#!/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 collections
import concurrent.futures
import os
import re
import sys

import lib.args
import lib.base
import lib.disk
import lib.lftest
import lib.lynis
import lib.net
import lib.shell
import lib.ssh
import lib.txt
from lib.globals import STATE_OK, STATE_UNKNOWN, STATE_WARN

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

DESCRIPTION = """Runs a lynis security audit across the hosts of a subnet or a host list from
a single management host, and reports each host's hardening index, its findings (the lynis
warnings) and the hardening suggestions of lynis. It discovers the targets (the subnet of the
default interface, a chosen interface or network, or an explicit host list), connects to each
one over SSH, copies a self-contained copy of lynis over, runs the audit there via password-less
sudo, retrieves the report and removes its temporary files. Alerts when a host is below the
hardening index threshold or reports a finding, and when the scan audited no host at all,
naming why the targets did not answer. The worst per-host result determines the overall state.
The scan runs as an unprivileged user and refuses to run as root. Security posture is
informational drift rather than a time-critical availability event, so by default only
WARNING is raised. The check is meant to run at most once per day. To evaluate the audit a
host runs on its own, use lynis-logfile on that host."""

DEFAULT_AUDIT_TIMEOUT = 600
DEFAULT_CONNECT_TIMEOUT = 3
DEFAULT_CRIT = ''
DEFAULT_MAX_WORKERS = 10
DEFAULT_NO_MATCH_SEVERITY = 'warn'
DEFAULT_PORT = 22
DEFAULT_WARN = '65:'

# Where lynis writes its output on a target when it runs privileged. The report and the
# log are left in place after the audit so an admin can inspect the details.
LYNIS_CUSTOM_PROFILE = lib.lynis.CUSTOM_PROFILE
LYNIS_LOG = lib.lynis.LOG
LYNIS_PID = lib.lynis.PID
LYNIS_REPORT = lib.lynis.REPORT

# Test IDs (`SSH-7408`, with an optional `:subtest` for skip-test), test groups and
# test categories. A value is written into a profile that lynis reads as root, so a
# newline must never get through: it would add a profile line of the caller's
# choice, `plugindir=` for example, which makes lynis source scripts from there.
# The patterns are applied with fullmatch(): `$` would still let a trailing newline
# through.
LYNIS_NAME_REGEX = re.compile(r'[A-Za-z0-9_-]+(?::[A-Za-z0-9_-]+)?')
# Auditor names and profile paths travel inside a double-quoted remote command in
# the network scan, and lynis splits the profile list on whitespace. From 3.1 on,
# lynis itself aborts on any argument with a character outside `[:space:]`,
# `a-zA-Z0-9` and `/._:=-` (SafeInput() in include/functions), a comma included.
LYNIS_AUDITOR_REGEX = re.compile(r'[A-Za-z0-9 ._-]+')
LYNIS_PROFILE_REGEX = re.compile(r'[A-Za-z0-9./_-]+')
# A raw option travels the same way, so it keeps to the characters lynis accepts
# itself, without `/`: no shell syntax, and no path. The options that take a path
# point lynis, running as root on the target, at binaries to run, scripts to source
# or files to write (include/parameters, lynis 3.1.8), and have no business here.
LYNIS_OPTION_REGEX = re.compile(r'-[A-Za-z0-9 ._:=-]*')
LYNIS_OPTIONS_WITH_PATH = (
    '--bindirs',
    '--bin-dirs',
    '--log-file',
    '--logfile',
    '--plugin-dir',
    '--plugindir',
    '--plugins-dir',
    '--profile',
    '--report-file',
    '--root-dir',
    '--rootdir',
    '--upload',
)

# Per-run remote work directory. lynis is a script that must be executed, so the
# directory has to live on a partition mounted without "noexec". Hardened hosts
# (exactly the ones being audited) usually mount /var/tmp and /tmp noexec, so the
# candidates are probed in order and the first one that can actually execute a
# file is used. The name is also used to validate the path before any removal.
REMOTE_WORKDIR_NAME = 'linuxfabrik-monitoring-plugins-lynis'
REMOTE_WORKDIR_BASES = ['/var/tmp', '/tmp', '"$HOME"', '/root', '/']  # nosec B108

# Where a packaged lynis keeps its plugins and default profile on the management
# host varies by distro (RHEL/Fedora/Arch ship the plugins under
# /usr/share/lynis, Debian/Ubuntu under /etc/lynis). These mirror lynis' own
# search order so the assembled copy is self-contained on every distro, which
# matters because a missing plugins directory aborts the remote audit.
LOCAL_PLUGIN_DIRS = (
    '/usr/local/lynis/plugins',
    '/usr/local/share/lynis/plugins',
    '/usr/share/lynis/plugins',
    '/etc/lynis/plugins',
)
LOCAL_PROFILE_FILES = (
    '/etc/lynis/default.prf',
    '/usr/share/lynis/default.prf',
)


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,
    )

    parser.add_argument(
        '--audit-timeout',
        help='Seconds to wait for the audit of a single host to finish. '
        'Default: %(default)s (seconds)',
        dest='AUDIT_TIMEOUT',
        type=int,
        default=DEFAULT_AUDIT_TIMEOUT,
    )

    parser.add_argument(
        '--configfile',
        help='SSH: Alternative per-user configuration file. If a configuration '
        'file is given on the command line, the system-wide '
        'configuration file (`/etc/ssh/ssh_config`) will be ignored. The '
        'default for the per-user configuration file is `~/.ssh/config`. If '
        'set to `none`, no configuration files will be read.',
        dest='CONFIGFILE',
    )

    parser.add_argument(
        '--connect-timeout',
        help='Seconds to wait for the SSH connection to a target host before '
        'skipping it. '
        'Default: %(default)s (seconds)',
        dest='CONNECT_TIMEOUT',
        type=int,
        default=DEFAULT_CONNECT_TIMEOUT,
    )

    # 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 per-host 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(
        '--disable-pseudo-terminal',
        help='SSH: Disable pseudo-terminal allocation.',
        dest='DISABLE_PSEUDO_TERMINAL',
        action='store_true',
        default=False,
    )

    parser.add_argument(
        '-H',
        '--host',
        help='Target host to audit. '
        'Can be specified multiple times. '
        'Takes precedence over --network and --interface. '
        'If none of the three is specified, the subnet of the default interface is '
        'scanned.',
        dest='HOST',
        action='append',
        default=None,
    )

    parser.add_argument(
        '--identity',
        help='SSH: File from which the identity (private key) for public '
        'key authentication is read. You can also specify a public key '
        'file to use the corresponding private key that is loaded in '
        'ssh-agent(1) when the private key file is not present locally. '
        'The default is `~/.ssh/id_dsa`, `~/.ssh/id_ecdsa`, '
        '`~/.ssh/id_ecdsa_sk`, `~/.ssh/id_ed25519`, `~/.ssh/id_ed25519_sk` and '
        '`~/.ssh/id_rsa`. Identity files may also be specified on a per-'
        'host basis in the configuration file. It is possible to have '
        'multiple --identity options (and multiple identities specified in configuration '
        'files). If no certificates have been explicitly specified by the '
        'CertificateFile directive, ssh will also try to load '
        'certificate information from the filename obtained by appending '
        '`-cert.pub` to identity filenames.',
        dest='IDENTITY',
        action='append',
    )

    parser.add_argument(
        '--interface',
        help='Network interface whose subnet is scanned. '
        'Ignored when --host or --network is given. '
        'If not specified, the default interface (the one carrying the default '
        'route) is used. '
        'Example: `--interface=eth0`',
        dest='INTERFACE',
        default=None,
    )

    parser.add_argument(
        '--ipv4',
        help='SSH: Forces ssh to use IPv4 addresses only.',
        dest='IPV4',
        action='store_true',
        default=False,
    )

    parser.add_argument(
        '--ipv6',
        help='SSH: Forces ssh to use IPv6 addresses only.',
        dest='IPV6',
        action='store_true',
        default=False,
    )

    # Deprecated: warnings and suggestions are always listed now.
    parser.add_argument(
        '--lengthy',
        help=argparse.SUPPRESS,
        dest='LENGTHY',
        action='store_true',
        default=False,
    )

    parser.add_argument(
        '--lynis-auditor',
        help='Name of the auditor to record in the report (lynis `--auditor`). '
        'If not specified, lynis uses its own default.',
        dest='LYNIS_AUDITOR',
        default=None,
    )

    parser.add_argument(
        '--lynis-option',
        help='Additional raw option to pass to the remote `lynis audit system` call, '
        'for options that have no dedicated parameter here. '
        'Options that take a path are not passed on, and an option may only contain '
        'letters, digits, spaces and `._:=-`. '
        'Can be specified multiple times. '
        'Example: `--lynis-option=--no-plugins`',
        dest='LYNIS_OPTION',
        action='append',
        default=None,
    )

    parser.add_argument(
        '--lynis-profile',
        help='Additional profile file for the audit (lynis `--profile`), read on '
        "top of lynis' own `default.prf` and `custom.prf`. "
        'The path is resolved on the target host. '
        'If not specified, lynis uses only its own profiles.',
        dest='LYNIS_PROFILE',
        default=None,
    )

    parser.add_argument(
        '--lynis-skip-test',
        help='Lynis test ID to skip (passed to lynis as `skip-test`), for '
        'exceptions controlled from the monitoring configuration. '
        'Can be specified multiple times. '
        "Host-specific exceptions belong in the host's own "
        '`/etc/lynis/custom.prf` instead. '
        'Example: `--lynis-skip-test=MAIL-8818`',
        dest='LYNIS_SKIP_TEST',
        action='append',
        default=None,
    )

    parser.add_argument(
        '--lynis-source',
        help='Path to a self-contained lynis directory (the '
        'directory that contains the `lynis` executable next to its `include`, '
        '`db` and `plugins` subdirectories). This is the copy that gets pushed to '
        'and run on every target host. '
        'If not specified, a self-contained copy is assembled from the local '
        'lynis installation.',
        dest='LYNIS_SOURCE',
        default=None,
    )

    parser.add_argument(
        '--lynis-test',
        help='Only run these lynis tests (lynis `--tests`). '
        'Can be specified multiple times. '
        'If not specified, all tests are run. '
        'Example: `--lynis-test=SSH-7408 --lynis-test=KRNL-5820`',
        dest='LYNIS_TEST',
        action='append',
        default=None,
    )

    parser.add_argument(
        '--lynis-test-category',
        help='Only run lynis tests of this category (lynis '
        '`--tests-from-category`). '
        'lynis runs one category at a time. '
        'If not specified, all categories are run. '
        'Example: `--lynis-test-category=security`',
        dest='LYNIS_TEST_CATEGORY',
        action='append',
        default=None,
    )

    parser.add_argument(
        '--lynis-test-group',
        help='Only run lynis tests of these groups (lynis `--tests-from-group`). '
        'Can be specified multiple times. '
        'If not specified, all groups are run. '
        'Example: `--lynis-test-group=ssh --lynis-test-group=kernel`',
        dest='LYNIS_TEST_GROUP',
        action='append',
        default=None,
    )

    parser.add_argument(
        '--max-workers',
        help='Maximum number of hosts to audit in parallel in the network scan. '
        'Default: %(default)s',
        dest='MAX_WORKERS',
        type=int,
        default=DEFAULT_MAX_WORKERS,
    )

    parser.add_argument(
        '--network',
        help='Network in CIDR notation whose hosts are audited. '
        'Can be specified multiple times. '
        'Takes precedence over --interface. '
        'Example: `--network=192.0.2.0/24`',
        dest='NETWORK',
        action='append',
        default=None,
    )

    parser.add_argument(
        '--no-match-severity',
        help=lib.args.help('--no-match-severity') + ' Default: %(default)s',
        dest='NO_MATCH_SEVERITY',
        choices=['ok', 'warn', 'crit', 'unknown'],
        default=DEFAULT_NO_MATCH_SEVERITY,
    )

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

    parser.add_argument(
        '-p',
        '--password',
        help='SSH: Password authentication. NOT RECOMMENDED. Requires `sshpass`. '
        'If you need to use password-based SSH login, run this plugin only on trusted hosts. '
        '`ps` will expose the SSH password.',
        dest='PASSWORD',
        default=None,
    )

    parser.add_argument(
        '--port',
        help='SSH: Port to connect to on the remote host. This can be specified on '
        'a per-host basis in the configuration file. Default: %(default)s',
        dest='PORT',
        default=DEFAULT_PORT,
    )

    parser.add_argument(
        '--quiet',
        help='SSH: Quiet mode. Causes most warning and diagnostic messages to be '
        'suppressed.',
        dest='QUIET',
        action='store_true',
        default=False,
    )

    parser.add_argument(
        '--ssh-option',
        help='SSH: Can be used to give options in the format used in the configuration file. '
        'This is useful for specifying options for which there '
        'is no separate command-line flag. For full details of the options, '
        'and their possible values, see ssh_config(5). Can be specified multiple times.',
        dest='SSH_OPTION',
        action='append',
    )

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

    parser.add_argument(
        '-u',
        '--username',
        help='SSH: Username. '
        'If not specified, ssh determines the user from `~/.ssh/config` or '
        'falls back to the current local user.',
        dest='USERNAME',
        default=None,
    )

    parser.add_argument(
        '--verbose',
        help=lib.args.help('--verbose'),
        dest='VERBOSE',
        action='store_true',
        default=False,
    )

    parser.add_argument(
        '-w',
        '--warning',
        help='WARN threshold for the per-host 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 validate_lynis_args(args):
    """Refuse values lynis would take for something else than the name they
    stand for. Returns (True, None) or (False, errormessage).
    """
    names = (
        ('--lynis-skip-test', args.LYNIS_SKIP_TEST),
        ('--lynis-test', args.LYNIS_TEST),
        ('--lynis-test-category', args.LYNIS_TEST_CATEGORY),
        ('--lynis-test-group', args.LYNIS_TEST_GROUP),
    )
    for param, values in names:
        for value in values:
            if not LYNIS_NAME_REGEX.fullmatch(value):
                return (
                    False,
                    f'{param} "{value}" is not a lynis test ID, group or category.',
                )
    # lynis compares the category for equality with the one of each test, so a
    # list would skip every test (include/functions, lynis 3.0.7 to 3.1.8)
    if len(args.LYNIS_TEST_CATEGORY) > 1:
        return (False, '--lynis-test-category can only be given once.')
    if args.LYNIS_AUDITOR and not LYNIS_AUDITOR_REGEX.fullmatch(args.LYNIS_AUDITOR):
        return (
            False,
            '--lynis-auditor may only contain letters, digits, spaces and `._-`.',
        )
    if args.LYNIS_PROFILE and not LYNIS_PROFILE_REGEX.fullmatch(args.LYNIS_PROFILE):
        return (
            False,
            '--lynis-profile may only contain letters, digits and `./_-`.',
        )
    for option in args.LYNIS_OPTION:
        if not LYNIS_OPTION_REGEX.fullmatch(option):
            return (
                False,
                f'--lynis-option "{option}" has to start with "-" and may only contain'
                ' letters, digits, spaces and `._:=-`.',
            )
        # The remote shell splits the value at spaces, so `--quick --profile x`
        # reaches lynis as three arguments: every word is checked, not only the
        # first one.
        for word in option.split():
            name = word.split('=', 1)[0]
            if name in LYNIS_OPTIONS_WITH_PATH:
                return (
                    False,
                    f'--lynis-option "{name}" takes a path and is not passed on;'
                    ' use the dedicated parameter where there is one.',
                )
    return (True, None)


def build_ssh_opts(args):
    """Assemble the common ssh/scp option string via lib.ssh, adding the
    non-interactive options this plugin needs (BatchMode, LogLevel,
    ConnectTimeout) on top of the by-ssh-compatible parameters.
    """
    return lib.ssh.build_options(
        configfile=args.CONFIGFILE,
        identity=args.IDENTITY,
        ssh_option=args.SSH_OPTION,
        ipv4=args.IPV4,
        ipv6=args.IPV6,
        quiet=args.QUIET,
        batch_mode=True,
        connect_timeout=args.CONNECT_TIMEOUT,
        log_level='ERROR',
    )


def ssh_exec(host, args, ssh_opts, remote, timeout=None):
    """Run `remote` on `host` over SSH (thin wrapper around lib.ssh.run that
    fills in this plugin's connection parameters).
    """
    return lib.ssh.run(
        host,
        remote,
        username=args.USERNAME,
        port=args.PORT,
        options=ssh_opts,
        disable_pseudo_terminal=args.DISABLE_PSEUDO_TERMINAL,
        password=args.PASSWORD,
        timeout=timeout,
    )


def scp_push(host, args, ssh_opts, local, remote, recursive=False):
    """Copy `local` to `remote` on `host` (thin wrapper around lib.ssh.scp)."""
    return lib.ssh.scp(
        host,
        local,
        remote,
        username=args.USERNAME,
        port=args.PORT,
        options=ssh_opts,
        password=args.PASSWORD,
        timeout=args.AUDIT_TIMEOUT,
        recursive=recursive,
    )


def push_lynis(host, args, ssh_opts, base_dir, remote_dir):
    """Copy the local lynis tree to `remote_dir`. Prefer rsync when it is
    available locally (faster for many files; run via sudo so the files land
    root-owned), and fall back to scp -r otherwise or when rsync fails (for
    example when the target has no rsync). Returns (success, (out, err, retc)).
    """
    if lib.shell.which('rsync'):
        lib.base.verbose(args.VERBOSE, f'{host}: pushing lynis via rsync...')
        success, result = lib.ssh.rsync(
            host,
            base_dir,
            remote_dir,
            username=args.USERNAME,
            port=args.PORT,
            options=ssh_opts,
            password=args.PASSWORD,
            timeout=args.AUDIT_TIMEOUT,
            sudo=True,
        )
        if success and result[2] == 0:
            return (True, result)
        lib.base.verbose(
            args.VERBOSE, f'{host}: rsync unavailable or failed, falling back to scp'
        )
    lib.base.verbose(args.VERBOSE, f'{host}: pushing lynis via scp...')
    return scp_push(host, args, ssh_opts, base_dir, remote_dir, recursive=True)


def build_lynis_command(args, base_dir):
    """Assemble the remote `./lynis audit system ...` invocation. --cronjob
    runs unattended (quick, quiet, no colors); --usecwd pins the pushed copy so
    a lynis installation already present on the target is ignored.
    """
    parts = ['./lynis audit system --cronjob --usecwd']
    # --plugin-dir is only pinned when the assembled copy actually ships a
    # plugins directory: pointing it at a missing directory makes lynis abort
    # before writing a report. The absolute path ($PWD is expanded on the
    # target, inside the extracted lynis directory) is needed because lynis
    # resolves a relative plugin directory against its working directory and
    # aborts once it changes directory during the scan. Without a plugins
    # directory the flag is omitted and lynis falls back to its own search.
    if lib.disk.dir_exists(os.path.join(base_dir, 'plugins')):
        parts[0] += ' --plugin-dir "$PWD/plugins"'
    if args.LYNIS_PROFILE:
        parts.append(f'--profile "{args.LYNIS_PROFILE}"')
    if args.LYNIS_AUDITOR:
        parts.append(f'--auditor "{args.LYNIS_AUDITOR}"')
    if args.LYNIS_TEST:
        parts.append(f'--tests "{" ".join(args.LYNIS_TEST)}"')
    if args.LYNIS_TEST_GROUP:
        parts.append(f'--tests-from-group "{" ".join(args.LYNIS_TEST_GROUP)}"')
    if args.LYNIS_TEST_CATEGORY:
        parts.append(f'--tests-from-category "{args.LYNIS_TEST_CATEGORY[0]}"')
    parts.extend(args.LYNIS_OPTION)
    return ' '.join(parts)


def connect_reason(text):
    """Reduce ssh's own error output for one target to a single comparable
    reason, so identical causes can be counted across targets. ssh prefixes the
    reason with its own name and with the target it tried ("ssh: connect to host
    192.0.2.1 port 22: Connection timed out") and prints a multi-line block
    before the decisive line (for example on a changed host key), so the prefix
    is dropped and the last meaningful line wins.
    """
    reason = ''
    for line in text.splitlines():
        line = line.strip()
        if not line or line.lower().startswith(('warning:', 'debug')):
            continue
        reason = line
    reason = re.sub(r'^ssh: (?:connect to host \S+ port \d+: )?', '', reason)
    return reason[:200] if reason else 'no answer on SSH'


def probe_host(host, args, ssh_opts):
    """Attempt an SSH connection to `host` and run a trivial command. Returns a
    (status, reason) tuple, where status is one of:
    - 'ok'   : connected and authenticated, ready to audit
    - 'auth' : host answered on SSH but authentication failed (wrong user/key);
               typical when auto-discovery probes a raw IP that has no matching
               `~/.ssh/config` host alias
    - 'down' : host did not answer (no SSH, filtered, or connect timeout)
    The reason carries ssh's own wording for a 'down' target, so an address that
    simply has no host behind it can be told apart from a target that is
    unreachable because of a configuration problem (changed host key,
    unresolvable name, missing `ssh`/`sshpass` binary, ...).
    """
    lib.base.verbose(args.VERBOSE, f'{host}: probing SSH connection...')
    success, result = ssh_exec(host, args, ssh_opts, 'true')
    if success and result[2] == 0:
        lib.base.verbose(args.VERBOSE, f'{host}: reachable')
        return ('ok', '')
    stderr = result[1] if success else str(result)
    if 'permission denied' in stderr.lower():
        lib.base.verbose(
            args.VERBOSE,
            f'{host}: SSH reachable but authentication failed, skipping audit',
        )
        return ('auth', '')
    reason = connect_reason(stderr)
    lib.base.verbose(args.VERBOSE, f'{host}: not reachable ({reason}), skipping')
    return ('down', reason)


def get_first_unreadable(path):
    """Return the first entry below `path` (or `path` itself) this account cannot
    read, or None if all of it can be read.
    """
    if os.path.isfile(path):
        return None if os.access(path, os.R_OK) else path
    if not os.access(path, os.R_OK | os.X_OK):
        return path
    for root, dirs, files in os.walk(path):
        for name in sorted(dirs):
            if not os.access(os.path.join(root, name), os.R_OK | os.X_OK):
                return os.path.join(root, name)
        for name in sorted(files):
            if not os.access(os.path.join(root, name), os.R_OK):
                return os.path.join(root, name)
    return None


def prepare_lynis_source(source):
    """Return (success, result). On success result is (base_dir, tmp_dir):
    `base_dir` is a self-contained lynis directory ready to be archived, and
    `tmp_dir` is a directory to clean up afterwards (or None if `base_dir`
    points at an existing tree that must not be removed).

    If `source` is given it is used as-is. Otherwise a self-contained copy is
    assembled from the local installation (`lynis` executable plus the
    `include`, `db`, `plugins` directories under /usr/share/lynis and the
    default profile), because a packaged install splits the executable from its
    data directories.
    """
    if source:
        if not lib.disk.file_exists(os.path.join(source, 'lynis')):
            return (
                False,
                f'--lynis-source "{source}" is not a self-contained lynis '
                'directory (no ./lynis executable found).',
            )
        return (True, (source, None))

    # autodetect a self-contained tree
    for candidate in ('/usr/local/lynis', '/opt/lynis'):
        if lib.disk.file_exists(
            os.path.join(candidate, 'lynis')
        ) and lib.disk.dir_exists(os.path.join(candidate, 'include')):
            return (True, (candidate, None))

    # assemble from a packaged installation
    executable = lib.shell.which('lynis')
    share = '/usr/share/lynis'
    if not executable or not lib.disk.dir_exists(os.path.join(share, 'include')):
        return (
            False,
            'No local lynis installation found. Install lynis or pass --lynis-source.',
        )
    # what goes into the copy, as (source, name in the copy)
    parts = [(executable, 'lynis')]
    for subdir in ('include', 'db'):
        src = os.path.join(share, subdir)
        if lib.disk.dir_exists(src):
            parts.append((src, subdir))
    # The plugins directory and the default profile live in different places
    # depending on the distro, so copy the first candidate found. Without a
    # plugins directory in the pushed copy, the remote audit aborts.
    for plugin_dir in LOCAL_PLUGIN_DIRS:
        if lib.disk.dir_exists(plugin_dir):
            parts.append((plugin_dir, 'plugins'))
            break
    for profile in LOCAL_PROFILE_FILES:
        if lib.disk.file_exists(profile):
            parts.append((profile, 'default.prf'))
            break

    # SUSE ships the files below /usr/share/lynis/include readable by root only
    # (0640, verified with lynis 3.0.8 on Leap 15.5 and 3.1.4 on Leap 16.0). Name the
    # first file this account cannot read instead of copying half a tree.
    for src, _ in parts:
        unreadable = get_first_unreadable(src)
        if unreadable:
            return (
                False,
                f'`{unreadable}` is not readable by this account, so the local lynis '
                'installation cannot be copied to the audited host. Unpack the lynis '
                'release tarball to /opt/lynis, which the check picks up by itself, or '
                'pass --lynis-source with another lynis tree this account can read.',
            )

    success, result = lib.disk.make_temp_dir(
        prefix='linuxfabrik-monitoring-plugins-lynis-'
    )
    if not success:
        return (False, result)
    tmp_dir = result
    base_dir = os.path.join(tmp_dir, 'lynis')
    success, error = lib.disk.mkdir(base_dir)
    for src, name in parts:
        if not success:
            break
        copy = lib.disk.copy_dir if os.path.isdir(src) else lib.disk.copy_file
        success, error = copy(src, os.path.join(base_dir, name))
    if not success:
        # the caller only cleans up what it got back
        lib.disk.rm_dir(tmp_dir)
        return (False, error)
    return (True, (base_dir, tmp_dir))


def extract_error(text):
    """Reduce noisy lynis output (SSH login banner, ASCII art, generic chatter)
    to the single most relevant error line, so the plugin output stays readable.
    Falls back to a generic message if nothing error-like is found.
    """
    error_regex = re.compile(
        r'(fatal|error|denied|cannot|no such|not found|timed out|timeout'
        r'|failed|sudo:|permission|quitting|symlink)',
        re.IGNORECASE,
    )
    candidates = [
        line.strip()
        for line in text.splitlines()
        if line.strip() and error_regex.search(line)
    ]
    if candidates:
        return candidates[-1][:200]
    return 'audit produced no report'


def audit_host(host, args, ssh_opts, base_dir, lynis_cmd):
    """Push a self-contained lynis copy to `host`, run a privileged audit, fetch
    the machine-readable report and remove the per-run work directory. The report
    and log are left on the host for the admin to inspect.
    Returns (success, report_text_or_errormessage).
    """
    short_timeout = args.CONNECT_TIMEOUT + 30

    # create a unique per-run work directory on the first candidate partition
    # that can actually execute a file (not mounted noexec). mktemp guarantees a
    # fresh, non-empty path; the path is echoed for the steps below and gets
    # validated before any recursive removal.
    # Best-practice preference order, then every mounted filesystem from
    # /proc/mounts as a fallback, so unusual layouts are still covered. The
    # first candidate that is both writable (mktemp succeeds) and executable
    # (the probe script runs) wins.
    bases = ' '.join(REMOTE_WORKDIR_BASES)
    make_workdir = (
        f'candidates="{bases}";'
        ' mounts=$(cut -d" " -f2 /proc/mounts 2>/dev/null);'
        ' for base in $candidates $mounts; do'
        f' dir=$(mktemp --directory "$base/{REMOTE_WORKDIR_NAME}.XXXXXXXX" 2>/dev/null)'
        ' || continue;'
        ' if printf "#!/bin/sh\\n" > "$dir/.exectest" && chmod +x "$dir/.exectest"'
        ' && "$dir/.exectest" >/dev/null 2>&1;'
        ' then rm --force "$dir/.exectest"; echo "$dir"; exit 0; fi;'
        ' rm --force "$dir/.exectest" 2>/dev/null; rmdir "$dir" 2>/dev/null;'
        ' done; exit 1'
    )
    success, result = ssh_exec(
        host, args, ssh_opts, make_workdir, timeout=short_timeout
    )
    if not success or result[2] != 0:
        return (
            False,
            'no executable (non-noexec) work directory found among '
            f'{", ".join(REMOTE_WORKDIR_BASES)}',
        )
    workdir = result[0].strip().splitlines()[-1].strip() if result[0].strip() else ''
    if not workdir.startswith('/') or f'/{REMOTE_WORKDIR_NAME}.' not in workdir:
        return (False, f'unexpected remote work directory: {workdir!r}')

    # push the self-contained lynis copy into the work directory (rsync if
    # available, else scp -r; neither needs `tar` on the target)
    success, result = push_lynis(host, args, ssh_opts, base_dir, f'{workdir}/lynis')
    if not success:
        return (False, f'push failed: {result}')
    _, push_stderr, push_retc = result
    if push_retc != 0:
        return (False, f'push failed: {push_stderr.strip() or f"retc {push_retc}"}')

    # run the audit. Single files are removed explicitly (never a recursive
    # wipe): the stale PID file would make lynis --cronjob refuse to run, and the
    # previous report/log are cleared so a failing run cannot leave a stale report
    # behind that looks like a fresh success. Because --usecwd makes lynis look
    # for profiles in the current directory only, the target's own custom.prf is
    # staged into the work directory so per-host suppressions (skip-test=...) are
    # honored. chown/chmod run via sudo so they work whether the files were
    # copied in root-owned (rsync --rsync-path) or user-owned (scp); chown also
    # silences lynis' ownership security check.
    # fleet-wide --lynis-skip-test entries are appended to the pushed default.prf
    # (double quotes only: the whole remote command is single-quoted by lib.ssh).
    # Host-specific skips come from the target's own custom.prf staged below;
    # lynis merges both profiles.
    skip_inject = ''
    if args.LYNIS_SKIP_TEST:
        skips = ' '.join(f'"{t}"' for t in args.LYNIS_SKIP_TEST)
        skip_inject = f' ; printf "skip-test=%s\\n" {skips} >> ./default.prf'

    remote_audit = (
        f'sudo --non-interactive rm --force {LYNIS_PID} {LYNIS_REPORT} {LYNIS_LOG}'
        f' ; cd "{workdir}/lynis" || exit 1'
        f' ; sudo --non-interactive cp --force {LYNIS_CUSTOM_PROFILE} ./custom.prf'
        ' 2>/dev/null'
        f'{skip_inject}'
        ' ; sudo --non-interactive chown --recursive 0:0 .'
        ' && sudo --non-interactive chmod +x ./lynis'
        f' && sudo --non-interactive {lynis_cmd}'
    )
    lib.base.verbose(
        args.VERBOSE, f'{host}: running audit (this can take a few minutes)...'
    )
    success, result = ssh_exec(
        host, args, ssh_opts, remote_audit, timeout=args.AUDIT_TIMEOUT
    )
    # lynis prints its abort reasons (for example a bad plugin directory) to
    # stdout, not stderr, so keep both streams to surface a useful message when
    # no report is written.
    audit_output = f'{result[0]}\n{result[1]}' if success else str(result)

    # fetch the report (root-owned, so cat it via sudo instead of scp)
    lib.base.verbose(args.VERBOSE, f'{host}: fetching report...')
    success, result = ssh_exec(
        host,
        args,
        ssh_opts,
        f'sudo --non-interactive cat {LYNIS_REPORT}',
        timeout=short_timeout,
    )
    report_text = result[0] if success else ''

    # remove the per-run work directory. The path comes from mktemp and was
    # validated above; the shell `case` guard repeats that check so a recursive
    # removal can only ever hit our own work directory, never an empty or
    # unexpected path. Report and log stay on the host for the admin.
    lib.base.verbose(args.VERBOSE, f'{host}: cleaning up...')
    ssh_exec(
        host,
        args,
        ssh_opts,
        f'case "{workdir}" in */{REMOTE_WORKDIR_NAME}.*)'
        f' sudo --non-interactive rm --recursive --force "{workdir}" ;; esac',
        timeout=short_timeout,
    )

    if 'hardening_index=' not in report_text:
        return (False, extract_error(audit_output))
    return (True, report_text)


def report_hostname(report):
    """Return the target's short hostname from its own lynis report (client-side,
    as the audited host sees itself).
    """
    return report['hostname']


def report_ip(report):
    """Return the target's primary, non-loopback IP address from its own lynis
    report (client-side). Prefers IPv4 over IPv6; empty string if none.
    """
    for ip in report['ipv4_addresses']:
        if not ip.startswith('127.'):
            return ip
    for ip in report['ipv6_addresses']:
        if ip != '::1' and not ip.lower().startswith('fe80'):
            return ip
    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)

    # set default values for append parameters that were not specified
    if args.LYNIS_OPTION is None:  # case 1: default to empty list
        args.LYNIS_OPTION = []
    if args.LYNIS_SKIP_TEST is None:  # case 1: default to empty list
        args.LYNIS_SKIP_TEST = []
    if args.LYNIS_TEST is None:  # case 1: default to empty list
        args.LYNIS_TEST = []
    if args.LYNIS_TEST_CATEGORY is None:  # case 1: default to empty list
        args.LYNIS_TEST_CATEGORY = []
    if args.LYNIS_TEST_GROUP is None:  # case 1: default to empty list
        args.LYNIS_TEST_GROUP = []
    # args.HOST, args.IDENTITY, args.INTERFACE, args.NETWORK, args.SSH_OPTION stay
    # None (case 3: None means "not set", which is different from an empty list)

    lib.base.coe(validate_lynis_args(args))

    # fetch data
    results = []
    targets = []
    if args.TEST is not None:
        # do not audit, put in test data: the fixture stands in for the report.dat
        # of one audited host
        stdout, _, _ = lib.lftest.test(args.TEST)
        targets = ['203.0.113.1']
        results = [('203.0.113.1', 'ok', stdout)]
    else:
        # As root, --configfile, --identity and --ssh-option would hand root to
        # whoever may run the check through sudo; a ProxyCommand is enough.
        if hasattr(os, 'geteuid') and os.geteuid() == 0:
            lib.base.oao(
                'The network scan does not run as root, because its SSH parameters '
                'would hand root to whoever may run the check. Run it as an '
                'unprivileged user, without sudo.',
                STATE_UNKNOWN,
                always_ok=args.ALWAYS_OK,
            )
        if args.HOST is not None:
            targets = args.HOST
        elif args.NETWORK is not None:
            lib.base.verbose(
                args.VERBOSE, f'Discovering hosts on {", ".join(args.NETWORK)}...'
            )
            targets = []
            for cidr in args.NETWORK:
                targets.extend(lib.base.coe(lib.net.cidr_to_hosts(cidr)))
        else:
            scope = args.INTERFACE if args.INTERFACE else 'the default interface'
            lib.base.verbose(args.VERBOSE, f'Discovering hosts on {scope}...')
            targets = lib.base.coe(lib.net.get_subnet_hosts(args.INTERFACE))
        lib.base.verbose(args.VERBOSE, f'{len(targets)} target(s) to check')
        ssh_opts = build_ssh_opts(args)

        lib.base.verbose(args.VERBOSE, 'Preparing local lynis copy...')
        base_dir, tmp_dir = lib.base.coe(prepare_lynis_source(args.LYNIS_SOURCE))
        lynis_cmd = build_lynis_command(args, base_dir)

        def process_host(host):
            """Probe and (if reachable and authenticated) audit one host.
            Returns a (host, outcome, result) tuple, where outcome is 'ok' for an
            audited host, 'failed' for a host that answered on SSH but could not
            be audited, and 'down' for an address that did not answer at all
            (result then carries ssh's reason).
            A failure on a single host must never abort the whole subnet scan,
            so any unexpected error is caught and reported as that host's
            result instead of propagating out of the worker pool.
            """
            try:
                status, reason = probe_host(host, args, ssh_opts)
                if status == 'down':
                    return (host, 'down', reason)
                if status == 'auth':
                    # host is up on SSH but we could not authenticate; surface
                    # it instead of silently skipping, so wrong credentials are
                    # visible
                    return (host, 'failed', 'SSH authentication failed')
                success, result = audit_host(host, args, ssh_opts, base_dir, lynis_cmd)
                if success:
                    report = lib.lynis.parse_report(result)
                    lib.base.verbose(
                        args.VERBOSE,
                        f'{host}: hardening index {report["hardening_index"]}',
                    )
                else:
                    lib.base.verbose(args.VERBOSE, f'{host}: audit failed ({result})')
                return (host, 'ok' if success else 'failed', result)
            except Exception as exc:  # isolate per-host failures from the scan
                lib.base.verbose(args.VERBOSE, f'{host}: audit failed ({exc})')
                return (host, 'failed', f'audit error: {exc}')

        try:
            # Audit hosts in parallel. map() preserves the input order, so the
            # output table is deterministic regardless of completion order.
            workers = max(1, min(args.MAX_WORKERS, len(targets))) if targets else 1
            with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
                results.extend(pool.map(process_host, targets))
        finally:
            if tmp_dir is not None and lib.disk.dir_exists(tmp_dir):
                lib.disk.rm_dir(tmp_dir)

    # init some vars
    msg = ''
    state = STATE_OK
    perfdata = ''
    table_data = []
    warning_rows = []
    suggestion_rows = []
    audited = 0
    reachable = 0
    unreachable = collections.Counter()
    total_warnings = 0
    total_suggestions = 0

    # analyze data
    # The state marker goes into the last table column on purpose: IcingaWeb
    # replaces "[WARNING]" with an icon and would otherwise break a monospace
    # table where the marked column is not the last one.
    for host, outcome, result in results:
        if outcome == 'down':
            # An address that does not answer on SSH is not a host, so it must
            # not show up as a failed one. The reason is counted instead, which
            # is what tells a dead address apart from a target this check cannot
            # reach because of a configuration problem.
            unreachable[result] += 1
            continue
        reachable += 1
        if outcome != 'ok':
            # no report, so no client-side hostname/IP; fall back to the target
            host_state = STATE_UNKNOWN
            state = lib.base.get_worst(state, host_state)
            table_data.append(
                {
                    'hostreport': f'{host}:-',
                    'ip': '-',
                    'warnings': '-',
                    'suggestions': '-',
                    'index': '-',
                    'state': f'{result}{lib.base.state2str(host_state, prefix=" ")}',
                }
            )
            continue
        audited += 1
        report = lib.lynis.parse_report(result)
        index = report['hardening_index']
        total_warnings += len(report['warnings'])
        total_suggestions += len(report['suggestions'])
        # hostname and IP come from the target's own report (client-side), not
        # from a lookup on the management host
        hostname = report_hostname(report) or host
        ip = report_ip(report) or '-'
        index_state = lib.base.get_state(index, args.WARN, args.CRIT, _operator='range')
        # 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.
        host_state = index_state
        if report['warnings']:
            host_state = lib.base.get_worst(host_state, STATE_WARN)
        state = lib.base.get_worst(state, host_state)
        for row in lib.lynis.parse_findings(report['warnings']):
            warning_rows.append({'host': hostname, **row})
        for row in lib.lynis.parse_findings(report['suggestions']):
            suggestion_rows.append({'host': hostname, **row})
        table_data.append(
            {
                'hostreport': f'{hostname}:{LYNIS_REPORT}',
                'ip': ip,
                'warnings': len(report['warnings']),
                'suggestions': len(report['suggestions']),
                'index': index,
                'state': lib.base.state2str(host_state),
            }
        )

    # build the message
    # The number of targets is the number of addresses probed, not the number of
    # hosts found: auto-discovery hands over every address of the subnet. Report
    # both, so a scan that found nothing is not mistaken for a scan of nothing.
    msg += f'{audited}/{reachable} {lib.txt.pluralize("host", reachable)} audited'
    msg += (
        f' ({len(targets)} {lib.txt.pluralize("address", len(targets), "es")} probed)'
    )
    if unreachable:
        # Cap the list: a subnet scan produces one entry per dead address, and the
        # reasons repeat.
        top = unreachable.most_common(3)
        listed = ', '.join(f'{count} x {reason}' for reason, count in top)
        rest = sum(unreachable.values()) - sum(count for _, count in top)
        if rest:
            listed += f', {rest} more'
        msg += f'\n\nNot reachable over SSH: {listed}'
    if not audited:
        # Nothing was audited, so a green result would only mean that nothing was
        # looked at.
        state = lib.base.get_worst(state, lib.base.str2state(args.NO_MATCH_SEVERITY))
    perfdata += lib.base.get_perfdata('hosts_total', len(targets), uom=None, _min=0)
    perfdata += lib.base.get_perfdata('hosts_reachable', reachable, uom=None, _min=0)
    perfdata += lib.base.get_perfdata('hosts_audited', audited, uom=None, _min=0)
    # every finding raises WARNING, so the line alerts from the first one on
    perfdata += lib.base.get_perfdata(
        'findings', total_warnings, uom=None, warn='0', _min=0
    )
    perfdata += lib.base.get_perfdata(
        'suggestions', total_suggestions, uom=None, _min=0
    )

    # build table output
    # State always sits in the last column: IcingaWeb replaces "[WARNING]" with
    # an icon and would otherwise break a monospace table.
    if table_data:
        keys = ['hostreport', 'ip', 'warnings', 'suggestions', 'index', 'state']
        headers = ['Host:Report', 'IP', 'Warn', 'Sugg', 'HIdx', 'State']
        msg += '\n\n' + lib.base.get_table(table_data, keys, header=headers).rstrip()
    # The findings themselves, each naming its host, so nobody has to collect the
    # reports by hand. A lynis warning is a finding and raises the state, a
    # suggestion is hardening advice and does not.
    msg += lib.lynis.get_findings_msg(warning_rows, suggestion_rows)

    # 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()
