#!/usr/bin/env python3
# -*- 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 sys

import lib.args
import lib.base
import lib.db_sqlite
import lib.lftest
import lib.logmatch
import lib.shell
import lib.time
import lib.txt
from lib.globals import STATE_CRIT, STATE_OK, STATE_UNKNOWN

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

DESCRIPTION = """Checks the kernel ring buffer (dmesg) for messages at severity levels emerg, alert,
crit, and err. Known false positives and hardware-specific noise are filtered out by default; the
filtered count is reported as the `errors` perfdata so trends can be graphed. Optionally asks
the monitoring server whether the service running this check is acknowledged: where it is, the
reported messages are suppressed on following runs so they don't re-alert, which makes a login
and "dmesg --clear" on the host unnecessary.
Note: the kernel ring buffer is a fixed-size circular buffer, so older messages are overwritten over
time, and timestamps may drift across SUSPEND/RESUME because the time source is not updated on
resume.
Requires root or sudo."""

ACK_RETENTION_DAYS = 30

DEFAULT_ICINGA_CALLBACK = False
DEFAULT_INSECURE = True
DEFAULT_NO_PROXY = False
DEFAULT_TIMEOUT = 5

# Ignore false positives, hardware-specific noise, and bugs without operational impact.
# Patterns are Python regular expressions matched against each dmesg line; keep them
# alphabetical and include the rationale so we can later re-evaluate whether an entry
# still applies. Only messages the kernel emits at emerg/alert/crit/err level can ever
# reach this filter, so verify the log level in the kernel source before adding an
# entry. Levels below err never show up here and an entry for them is dead weight.
# The level differs per distribution, so check the oldest one we build for, not the
# current upstream kernel. The entries below were verified against RHEL 8 (4.18.0-553),
# RHEL 9 (5.14.0-687), Debian 11 (5.10), SLES 15 (5.14), Ubuntu 22.04 (5.15),
# Debian 12 (6.1), Ubuntu 24.04 (6.8), Debian 13 / Rocky 10 / SLES 16 (6.12) and 7.1.
DEFAULT_IGNORE = [
    # ACPI method aborted because it references an object the firmware never declared.
    # ACPICA treats namespace lookup failures as firmware bugs; aborts caused by other
    # AML errors keep being reported.
    r' ACPI BIOS Error \(bug\): Could not resolve symbol \[.*\], AE_NOT_FOUND',
    r' ACPI BIOS Error \(bug\): Failure creating named object \[.*\], AE_ALREADY_EXISTS',
    r' ACPI Error: Aborting method .* due to previous error \(AE_NOT_FOUND\)',
    # RTC wake alarm cannot be armed; only wake-from-RTC is affected, timekeeping is not
    ' ACPI Error: Could not enable RealTimeClock event',
    # SCSI sd: cache mode page absent (Virtio/USB/SD-card disks); falls back to
    # write-through. `Assuming drive cache` is err level on Debian 11, SLES 15 and
    # Ubuntu 22.04 only; RHEL and Linux 5.16 and later log it as a warning
    ' Asking for cache data failed',
    ' Assuming drive cache: write through',
    # Broadcom WLAN firmware-load info on Raspberry Pi 3B+ (BCM4345/6). The Raspberry Pi
    # kernel is built with CONFIG_BRCMDBG=y, which promotes these info lines to err level
    ' brcmfmac: brcmf_c_preinit_dcmds: Firmware: BCM4345/6',
    ' brcmfmac: brcmf_fw_alloc_request: using brcm/brcmfmac43455-sdio'
    ' for chip BCM4345/6',
    # KVM guest sets branch-trace/last-branch bits the host emulates as a no-op;
    # typically Windows guests on KVM/oVirt
    r' BTF\|LBR in IA32_DEBUGCTLMSR ',
    # KVM guest perfctr writes blocked, https://access.redhat.com/solutions/2188061
    ' disabled perfctr wrmsr',
    # EDAC probes an Intel memory controller the board or hypervisor does not expose.
    # Without a controller there is nothing to report ECC errors from. Intel EDAC drivers
    # skip virtual machines since Linux 5.14, older kernels log this in every guest
    ' EDAC i10nm: No memory controllers found',
    r' EDAC (i7core|pnd2|sbridge): Failed to register device with error ',
    # Shim/MOK config table not exposed as EFI runtime memory; cosmetic, no Secure Boot
    # impact. RHEL 8 only, later kernels reworded the check and log it as a warning.
    # Documented for Rocky Linux 8.5, https://rockylinux.org/news/rocky-linux-8-5-ga-release/
    ' EFI MOKvar config table is not in EFI runtime memory',
    # ACPI Error Record Serialization Table not provided by firmware; common on most boards/VMs
    r' ERST: Failed to get Error Log Address Range\.',
    # AMD: firmware already claimed the local APIC vector the kernel wants for IBS or for
    # the MCE threshold interrupt. Those two features stay off, nothing else is affected
    r' \[Firmware Bug\]: cpu \d+, failed to setup threshold interrupt',
    r' \[Firmware Bug\]: cpu \d+, IBS interrupt offset \d+ not available',
    r' \[Firmware Bug\]: cpu \d+, invalid IBS interrupt offset',
    r' \[Firmware Bug\]: cpu \d+, try to use APIC[0-9A-F]+ \(LVT offset \d+\)',
    # BIOS keeps the hardware performance counters for itself; only perf-based tooling
    # is affected, common on servers that expose power or thermal monitoring to the BMC
    r' \[Firmware Bug\]: the BIOS has corrupted hw-PMU resources',
    # TSC deadline timer disabled by a CPU erratum; kernel falls back to the APIC timer
    r' \[Firmware Bug\]: TSC_DEADLINE disabled due to Errata',
    # DRM vsync flip timeout on i915 / virt. GPUs, https://access.redhat.com/solutions/4490391
    ' flip_done timed out',
    # Emulated floppy or empty optical drive in a VM answers every read with an error.
    # Scoped to fd*/sr* so real block device errors keep being reported
    r' (I/O error, dev|Buffer I/O error on dev(ice)?) (fd|sr)\d',
    # IMA/EVM cannot load kernel-shipped X.509 cert (-126 ENOKEY); MOK keyring not yet populated,
    # https://access.redhat.com/solutions/7049158
    r' integrity: Problem loading X\.509 certificate -126',
    # BMC interface advertised via DMI/SMBIOS is not usable; ipmi_si moves on to the next
    # interface source. Common on boards whose firmware announces a BMC that is not fitted
    r' ipmi_si .*: (Could not set up I/O space|Interface detection failed)',
    # SCSI sd: same probe path as "Asking for cache data failed", same level history
    ' No Caching mode page found',
    # SHPC PCI hot-plug slot already owned by acpiphp/pciehp on virt. PCI bridges (-16 EBUSY);
    # hot-plug keeps working via the other driver. Common on OpenStack/KVM/VMware guests.
    # Anchored on shpchp so that pciehp slot failures on real hardware keep being reported
    ' pci_hp_register failed with error -16',
    r' shpchp .*: Slot initialization failed',
    # SMBus controller absent or BIOS-disabled (i2c-piix4 / i2c-i801 / i2c-ismt); no impact
    # on monitoring, https://access.redhat.com/solutions/2115401
    ' SMBus base address uninitialized, upgrade BIOS',
    ' SMBus Host Controller not enabled!',
    # KVM guest reads unhandled MSR, https://access.redhat.com/solutions/59299
    ' unhandled rdmsr: ',
    # KVM guest writes unhandled MSR, https://bugzilla.redhat.com/show_bug.cgi?id=874627
    ' unhandled wrmsr: ',
    # RHEL flags driver as deprecated/unmaintained for the next major release; informational only
    ' Warning: Deprecated Driver is detected',
    ' Warning: Unmaintained driver is detected',
    # SGX turned off in firmware; the kernel just clears the CPU feature. RHEL 8 words this
    # as "SGX disabled by BIOS."
    ' x86/cpu: SGX disabled (or unsupported )?by BIOS',
]


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(
        '--icinga-callback',
        help=lib.args.help('--icinga-callback'),
        dest='ICINGA_CALLBACK',
        action='store_true',
        default=DEFAULT_ICINGA_CALLBACK,
    )

    parser.add_argument(
        '--icinga-password',
        help=lib.args.help('--icinga-password'),
        dest='ICINGA_PASSWORD',
    )

    parser.add_argument(
        '--icinga-service-name',
        help=lib.args.help('--icinga-service-name'),
        dest='ICINGA_SERVICE_NAME',
    )

    parser.add_argument(
        '--icinga-url',
        help=lib.args.help('--icinga-url'),
        dest='ICINGA_URL',
    )

    parser.add_argument(
        '--icinga-username',
        help=lib.args.help('--icinga-username'),
        dest='ICINGA_USERNAME',
    )

    # Append parameters use `default=None`; the actual default list (`DEFAULT_IGNORE`)
    # is assigned in main() if the user did not pass `--ignore`. Specifying `--ignore`
    # at least once therefore replaces the default list rather than extending it; this
    # matches the convention documented in CONTRIBUTING.md and lets admins curate
    # their own ignore list without inheriting the bundled defaults.
    parser.add_argument(
        '--ignore',
        help='Ignore a kernel message matching this Python regular expression. '
        'Can be specified multiple times. '
        'Specifying this parameter replaces the bundled default ignore list. '
        'Example: `--ignore="^.* unhandled (rd|wr)msr: "`.',
        dest='IGNORE',
        action='append',
        default=None,
    )

    parser.add_argument(
        '--insecure',
        help='Applies to the connection to the monitoring server that `--icinga-callback` makes, which is the only network connection this check opens. '
        + lib.args.help('--insecure'),
        dest='INSECURE',
        action='store_true',
        default=DEFAULT_INSECURE,
    )

    parser.add_argument(
        '--match',
        help='Only consider a kernel message matching this Python regular expression. '
        'Case-sensitive by default; use `(?i)` for case-insensitive matching. '
        'Can be specified multiple times. '
        + lib.args.MATCH_IGNORE_PRECEDENCE
        + " Example: `--match='blk_update_request'`.",
        dest='MATCH',
        action='append',
        default=None,
    )

    parser.add_argument(
        '--no-insecure',
        help='Applies to the connection to the monitoring server that `--icinga-callback` makes, which is the only network connection this check opens. '
        + lib.args.help('--no-insecure'),
        dest='INSECURE',
        action='store_false',
        default=DEFAULT_INSECURE,
    )

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

    parser.add_argument(
        '--no-proxy',
        help='Applies to the connection to the monitoring server that `--icinga-callback` makes, which is the only network connection this check opens. '
        + lib.args.help('--no-proxy'),
        dest='NO_PROXY',
        action='store_true',
        default=DEFAULT_NO_PROXY,
    )

    # `--severity` is no longer exposed: kernel ring buffer messages on err level
    # are not a meaningful "warning" in a server-hosting context, so the plugin
    # always alerts as CRIT. Kept hidden via SUPPRESS for backwards compatibility
    # with existing service templates.
    parser.add_argument(
        '--proxy',
        help='Applies to the connection to the monitoring server that `--icinga-callback` makes, which is the only network connection this check opens. '
        + lib.args.help('--proxy'),
        dest='PROXY',
        default=None,
    )

    parser.add_argument(
        '--severity',
        help=argparse.SUPPRESS,
        dest='SEVERITY',
    )

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

    parser.add_argument(
        '--timeout',
        help=lib.args.help('--timeout') + ' Default: %(default)s (seconds)',
        dest='TIMEOUT',
        type=int,
        default=DEFAULT_TIMEOUT,
    )

    args, _ = parser.parse_known_args()
    return args


def get_line_fingerprint(line):
    """Return a stable fingerprint for a single kernel message.

    Keys on the complete dmesg line including its timestamp, so the very same
    message logged again later counts as a new event and alerts again.
    """
    return lib.logmatch.key(line.strip())


def migrate_ack_state(conn):
    """Carry acknowledgements written by an earlier version over to the current layout.

    Without this, the first run after an update finds no acknowledgement at all and raises
    every message an operator had already taken on, on every host at once.

    The migration works inside the state database that is already open, and never constructs,
    moves or opens a path of its own. Keep it that way: the previous attempt at carrying old
    state over moved a file from a predictable location in the shared temp directory and
    followed a symlink planted there (GHSA-w2gg-hx6w-24w3).
    """
    tables = lib.base.coe(lib.db_sqlite.get_tables(conn))
    if 'acknowledged_lines' not in tables:
        return
    rows = lib.base.coe(
        lib.db_sqlite.select(
            conn,
            'SELECT line_hash FROM acknowledged_lines',
            fetchone=False,
        )
    )
    if rows:
        # The line itself was never stored in the old layout, only its hash, so
        # the text stays empty here. It is not read back anywhere: the hash is
        # what an acknowledgement is matched on.
        lib.base.coe(
            lib.logmatch.acknowledge(
                conn,
                [{'key': row['line_hash'], 'line': ''} for row in rows],
            )
        )
    lib.base.coe(lib.db_sqlite.drop_table(conn, table='acknowledged_lines'))


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)

    # apply default ignore list if the admin did not pass --ignore
    if args.IGNORE is None:
        args.IGNORE = DEFAULT_IGNORE
    if args.MATCH is None:
        args.MATCH = []

    if args.ICINGA_CALLBACK and not all(
        (
            args.ICINGA_URL,
            args.ICINGA_PASSWORD,
            args.ICINGA_USERNAME,
            args.ICINGA_SERVICE_NAME,
        )
    ):
        lib.base.cu(
            '`--icinga-callback` requires `--icinga-url`, `--icinga-password`, `--icinga-username` and `--icinga-service-name`.'
        )

    # compile ignore patterns (one coe per item so we get a per-pattern error message)
    ignore_patterns = [
        lib.base.coe(lib.txt.compile_regex(p, key='--ignore')) for p in args.IGNORE
    ]

    # Persisted ack state is only needed when the Icinga callback is in use.
    # When it is, each ignore list gets its own state DB so two Icinga services
    # watching the ring buffer with different filters do not share ack state.
    acked_fingerprints = set()
    ack_conn = None
    if args.ICINGA_CALLBACK:
        instance = lib.logmatch.instance_id({'ignore': args.IGNORE})
        ack_conn = lib.base.coe(lib.logmatch.connect('dmesg', instance))
        migrate_ack_state(ack_conn)
        # Drop ack records older than the retention: by that age the message has
        # been overwritten in the ring buffer and can no longer re-appear anyway.
        lib.base.coe(lib.logmatch.prune(ack_conn, retention=ACK_RETENTION_DAYS))
        acked_fingerprints = lib.base.coe(lib.logmatch.suppressed(ack_conn))

    # fetch data
    if args.TEST is None:
        stdout, stderr, retc = lib.base.coe(
            lib.shell.shell_exec(['dmesg', '--level=emerg,alert,crit,err', '--ctime']),
        )
        if stderr or retc != 0:
            # A kernel with `kernel.dmesg_restrict=1`, the default on many
            # distributions, refuses the read for anyone but root. Naming the reason
            # beats handing over "read kernel buffer failed: Operation not permitted",
            # which says what failed but not what to do about it.
            if 'operation not permitted' in stderr.lower():
                lib.base.cu(
                    'Not allowed to read the kernel log. Run this plugin as root or '
                    'via sudo, or set `kernel.dmesg_restrict=0`.'
                )
            lib.base.cu(stderr)
    else:
        stdout, stderr, retc = lib.lftest.test(args.TEST)

    # init some vars
    msg = ''
    state = STATE_OK
    perfdata = ''

    # analyze data: filter the dmesg output line by line
    all_lines = [line for line in stdout.strip().split('\n') if line]
    match_patterns = [
        lib.base.coe(item) for item in lib.txt.compile_regex(args.MATCH, '--match')
    ]
    # `--match` (include) is applied first, then `--ignore` (exclude), so a
    # message hit by `--ignore` is dropped even if it also matches `--match`.
    errors = [
        line
        for line in all_lines
        if (not match_patterns or any(p.search(line) for p in match_patterns))
        and not any(p.search(line) for p in ignore_patterns)
    ]
    # drop the messages that were acknowledged on an earlier run
    suppressed_cnt = 0
    if acked_fingerprints:
        remaining = [
            line
            for line in errors
            if get_line_fingerprint(line) not in acked_fingerprints
        ]
        suppressed_cnt = len(errors) - len(remaining)
        errors = remaining
    cnt = len(errors)
    if cnt > 0:
        state = STATE_CRIT

    # Ask Icinga about the service acknowledgement. If acknowledged, persist the
    # fingerprints of the messages that are currently being reported so they do
    # not re-alert on following runs, and return OK to Icinga. See issue #639.
    msg_addendum = ''
    if args.ICINGA_CALLBACK and state != STATE_OK:
        acknowledged, msg_addendum = lib.base.coe(
            lib.logmatch.service_acknowledged(
                args.ICINGA_URL,
                args.ICINGA_USERNAME,
                args.ICINGA_PASSWORD,
                args.ICINGA_SERVICE_NAME,
                insecure=args.INSECURE,
                no_proxy=args.NO_PROXY,
                proxy=args.PROXY,
                timeout=args.TIMEOUT,
            )
        )
        if acknowledged:
            # Persist the messages currently being reported, so an unchanged
            # ring buffer does not raise them again on the next run (issue
            # #639). The key is derived from the line, so the very same message
            # logged again later is a new event and alerts again.
            lib.base.coe(
                lib.logmatch.acknowledge(
                    ack_conn,
                    [
                        {'key': get_line_fingerprint(line), 'line': line}
                        for line in errors
                    ],
                )
            )
            state = STATE_OK

    if ack_conn is not None:
        lib.db_sqlite.close(ack_conn)

    # build the message
    suppressed_msg = ''
    if suppressed_cnt > 0:
        suppressed_msg = (
            f' {suppressed_cnt} acknowledged'
            f' {lib.txt.pluralize("message", suppressed_cnt)} suppressed.'
        )
    if cnt > 0:
        # shorten the message to first 5 and last 5 lines if it gets large
        shown = [*errors[0:5], '...', *errors[-5:]] if cnt > 10 else errors
        msg += (
            f'{cnt} {lib.txt.pluralize("error", cnt)} in Kernel Ring Buffer.'
            f'{suppressed_msg}\n\n'
            + '\n'.join(
                shown,
            )
        )
    else:
        msg += f'Everything is ok.{suppressed_msg}'
    if msg_addendum:
        msg += '\n\n' + msg_addendum

    # build perfdata
    perfdata += lib.base.get_perfdata('errors', cnt, _min=0)

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