#!/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.human
import lib.lftest
import lib.shell
import lib.txt
from lib.globals import STATE_OK, STATE_UNKNOWN

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

DESCRIPTION = """Checks for available APT package updates on Debian, Ubuntu, and compatible
systems. Reports the number of pending updates and how many of them come from a security
repository. Alerts when the number of pending updates reaches the warning threshold, and
when the number of security updates reaches the critical threshold. This check only lists
updates and never actually installs anything. Requires root or sudo."""

DEFAULT_CRIT = None  # number of security updates, off by default
# `apt list --upgradable` prints "<package>/<suite>[,<suite>...] <version> <arch> [...]".
# Anchoring on the suite field catches a package that is offered by the security archive
# alone ("aom-tools/stable-security"), which is what a fresh CVE fix looks like before the
# next point release folds it into the main suite, as well as one offered by both
# ("gzip/noble-updates,noble-security"). Verified against apt 2.6 on Debian 12/13 and apt
# 2.7 on Ubuntu 24.04.
DEFAULT_CRITICAL_PATTERN = r'^\S+/\S*-security'
DEFAULT_GRACE_SECURITY = '0D'
DEFAULT_GRACE_UPDATES = '0D'
DEFAULT_QUERY = '1'
DEFAULT_TIMEOUT = 60
DEFAULT_WARN = 1  # number of updatable packages


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(
        '-c',
        '--critical',
        help='Minimum number of pending security updates to trigger a CRITICAL. '
        'Counts the updates matching `--critical-pattern`, within the scope of '
        '`--query`. '
        'Unset by default, so security updates raise a WARNING like any other '
        'update until a threshold is given. '
        'Example: `--critical=1` '
        'Default: %(default)s',
        dest='CRIT',
        type=int,
        default=DEFAULT_CRIT,
    )

    parser.add_argument(
        '--critical-pattern',
        help='Marks an update as security-critical. '
        'Matched against the whole line as printed by `apt list --upgradable`, so it '
        'can key on the package name, the suite, or the version. '
        'Uses Python regular expressions. '
        'Case-sensitive. '
        "Example: `--critical-pattern='^\\S+/\\S*(-security|-lts)'` "
        'Default: %(default)s',
        dest='CRITICAL_PATTERN',
        default=DEFAULT_CRITICAL_PATTERN,
    )

    parser.add_argument(
        '--grace-security',
        help=lib.args.help('--grace-security') + ' Default: %(default)s',
        dest='GRACE_SECURITY',
        type=lib.args.duration,
        default=DEFAULT_GRACE_SECURITY,
    )

    parser.add_argument(
        '--grace-updates',
        help=lib.args.help('--grace-updates') + ' Default: %(default)s',
        dest='GRACE_UPDATES',
        type=lib.args.duration,
        default=DEFAULT_GRACE_UPDATES,
    )

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

    parser.add_argument(
        '--no-update',
        help='Skip the package cache refresh and evaluate the cache as it is. '
        'Without this, the check runs `apt-get update` first, which needs sudo and '
        'reaches the package repositories on every run. '
        'Use it on hosts where a timer already keeps the cache fresh.',
        dest='NO_UPDATE',
        action='store_true',
        default=False,
    )

    parser.add_argument(
        '--only-critical',
        help='Only report security-critical updates and upgrades.',
        dest='ONLY_CRITICAL',
        action='store_true',
        default=False,
    )

    parser.add_argument(
        '--query',
        help='SQL WHERE clause to narrow down results from the internal updates table. '
        'Supports regular expressions via a REGEXP statement. '
        'If specified, a list of matching updates is printed. '
        'Have a look at the README for a list of available columns. '
        'Example: `--query=\'package like "bind9-%%"\'`. '
        'Default: %(default)s',
        dest='QUERY',
        default=DEFAULT_QUERY,
    )

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

    parser.add_argument(
        '-w',
        '--warning',
        help='Minimum number of pending updates to trigger a WARNING. '
        'Default: %(default)s',
        dest='WARN',
        type=int,
        default=DEFAULT_WARN,
    )

    args, _ = parser.parse_known_args()
    return args


def package_name(line):
    """Return the package name of an `apt list --upgradable` line, which prints
    `<package>/<suite>[,<suite>...] <version> <arch> [...]`.

    Ageing an update keys on the name alone. Keying on the version would restart
    the clock on every rebuild in the repository, which on a fast-moving suite
    means a package could stay inside its grace period forever.
    """
    return line.split('/', 1)[0]


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 args.NO_UPDATE:
            # try to update the package cache first
            _, stderr, retc = lib.base.coe(
                lib.shell.shell_exec(
                    ['sudo', 'apt-get', 'update', '--quiet', '2'], timeout=args.TIMEOUT
                ),
            )
            if retc or stderr:
                # Not printing stderr as it is quite verbose
                lib.base.cu(
                    '`apt-get update` returned with an error. '
                    'Check the sudo permissions, or use --no-update to evaluate the '
                    'package cache without refreshing it.'
                )
        stdout, _, retc = lib.base.coe(
            lib.shell.shell_exec(['apt', 'list', '--upgradable'], timeout=args.TIMEOUT),
        )
    else:
        # skip the apt-get update refresh in test mode and read the
        # `apt list --upgradable` output from a fixture file
        stdout, _, retc = lib.lftest.test(args.TEST)
    if retc:
        # Not printing stderr as it is quite verbose
        lib.base.cu('`apt list --upgradable` returned with an error.')

    # init some vars
    msg = ''
    state = STATE_OK
    perfdata = ''
    critical_pattern = lib.base.coe(
        lib.txt.compile_regex(args.CRITICAL_PATTERN, '--critical-pattern')
    )

    # strip first line "Listing... Done." (if any)
    first, _, rest = stdout.partition('\n')
    stdout = rest if first.lower().startswith('listing...') else stdout
    upgradable = stdout.strip().splitlines()

    # Remember since when an update is pending for each package, so the grace
    # periods below can hold an alert back until the host has had a patch window.
    # Ages are kept for every upgradable package, not for the subset `--query`
    # selects, so two services with different queries do not forget each other's
    # packages. Skipped under `--test`, which must not touch the host's state.
    ages = None
    if args.TEST is None:
        ages = lib.db_sqlite.first_seen(
            'linuxfabrik-monitoring-plugins-deb-updates.db',
            'deb-updates',
            [package_name(line) for line in upgradable],
        )

    # Create the db table. Apart from the security flag this is just one column,
    # cause *apt does not have a stable CLI interface*. The database is private to
    # this process: it holds nothing worth keeping between runs, and a file under a
    # fixed name is shared with every concurrent run of this check, which then count
    # each other's rows.
    definition = """
        package  TEXT DEFAULT NULL,
        critical INT  DEFAULT 0
    """
    conn = lib.base.coe(lib.db_sqlite.connect(in_memory=True))
    lib.base.coe(
        lib.db_sqlite.create_table(
            conn,
            definition,
            table='deb_updates',
        )
    )

    # analyze data
    for item in upgradable:
        lib.base.coe(
            lib.db_sqlite.insert(
                conn,
                {
                    'package': item,
                    'critical': 1 if critical_pattern.search(item) else 0,
                },
                table='deb_updates',
            )
        )

    # store table_data in local sqlite database
    lib.base.coe(lib.db_sqlite.commit(conn))

    # fetch desired objects only, and set sqlite3 to be case insensitive when string comparing
    # QUERY is by design an admin-provided SQL WHERE clause (documented feature)
    sql = f"""
        SELECT *
        FROM deb_updates
        WHERE {args.QUERY}
        COLLATE NOCASE
    """  # nosec B608
    result = lib.base.coe(lib.db_sqlite.select(conn, sql))
    lib.db_sqlite.close(conn)

    # `--only-critical` narrows down what is reported, while the security count
    # itself stays available either way, so both numbers can be trended at once
    critical_count = sum(1 for row in result if row['critical'])
    if args.ONLY_CRITICAL:
        result = [row for row in result if row['critical']]

    # Only updates that outlived their grace period drive the state. An age we
    # could not read counts as old enough, so a cache that went missing never
    # silences the check. The reported and graphed numbers stay untouched: what
    # is pending is pending, the grace period only postpones the alert.
    def grace_of(row):
        # A security update is held back by `--grace-security`, everything else
        # by `--grace-updates`. Both counts share this rule, so the default
        # `--grace-security=0D` keeps security updates alerting right away even
        # when `--critical` is unset and they are only counted as ordinary updates.
        return args.GRACE_SECURITY if row['critical'] else args.GRACE_UPDATES

    def is_due(row):
        if ages is None:
            return True
        return ages.get(package_name(row['package']), 0) >= grace_of(row)

    due_count = sum(1 for row in result if is_due(row))
    due_critical_count = sum(1 for row in result if row['critical'] and is_due(row))
    within_grace = len(result) - due_count

    # get state
    updates_state = lib.base.get_state(due_count, args.WARN, None)
    critical_state = lib.base.get_state(due_critical_count, None, args.CRIT)
    state = lib.base.get_worst(updates_state, critical_state)

    def row_state(row):
        # The sentence above the list only carries totals, and the number that
        # drives the state is the number of due updates, not the number
        # available. Naming per package the state it feeds traces the check's
        # own state back to the packages responsible for it, and says of the
        # rest when they follow. Both thresholds count packages, so no single
        # package crosses one on its own: every due one carries the state its
        # count produced. A package that alerts nobody carries no marker rather
        # than an `[OK]`, which next to a pending update looks like there is
        # nothing left to do; it still says `overdue`, so every package that
        # has run out its grace period reads the same way.
        if not is_due(row):
            left = grace_of(row) - ages.get(package_name(row['package']), 0)
            return f'due in {lib.human.seconds2human(left)}'
        worst = updates_state
        if row['critical']:
            worst = lib.base.get_worst(worst, critical_state)
        return f'overdue{lib.base.state2str(worst, prefix=" ")}'

    # build the message
    query_hint = f' (query: {args.QUERY})' if args.QUERY != DEFAULT_QUERY else ''
    if len(result) == 0:
        msg += f'No updates available{query_hint}.'
    else:
        msg += f'{len(result)} {"critical " if args.ONLY_CRITICAL else ""}'
        msg += f'{lib.txt.pluralize("update", len(result))} available'
        if critical_count and not args.ONLY_CRITICAL:
            msg += f', {critical_count} of them critical'
        msg += query_hint
        msg += '.'
        if within_grace:
            # Naming the periods answers the question the sentence otherwise
            # provokes: an admin who reads that something is being held back wants
            # to know for how long, and the two are not always the same.
            msg += (
                f' {within_grace} of them within the grace period'
                f' (updates: {args.GRACE_UPDATES}, security: {args.GRACE_SECURITY}).'
            )
        msg += lib.base.state2str(state, prefix=' ')
        msg += '\n* '
        msg += '\n* '.join(
            [f'{row["package"]} {row_state(row)}'.rstrip() for row in result]
        )
    perfdata += lib.base.get_perfdata(
        'updates',
        len(result),
        uom=None,
        warn=args.WARN,
        _min=0,
    )
    perfdata += lib.base.get_perfdata(
        'critical_updates',
        critical_count,
        uom=None,
        crit=args.CRIT,
        _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()
