#!/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 re
import sys

import lib.args
import lib.base
import lib.lftest
import lib.txt
from lib.globals import STATE_OK, STATE_UNKNOWN

try:
    import psutil
except ImportError:
    print('Python module "psutil" is not installed.')
    sys.exit(STATE_UNKNOWN)


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

DESCRIPTION = """Checks the state of one or more Windows services. Accepts the case-insensitive
service name (not the display name) and supports regular expressions to match multiple
services. Alerts on services that are not in the expected state."""

DEFAULT_CRIT = None
DEFAULT_SEVERITY = 'warn'
DEFAULT_STATUS = ['running']
DEFAULT_STARTTYPE = ['automatic']
DEFAULT_WARN = '1:'

STARTTYPES = ['automatic', 'disabled', 'manual']
STATUSES = [
    'continue_pending',
    'pause_pending',
    'paused',
    'running',
    'start_pending',
    'stop_pending',
    'stopped',
]


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='CRIT threshold for the number of services in the expected status. '
        'Supports Nagios ranges. '
        'Default: %(default)s',
        dest='CRIT',
        default=DEFAULT_CRIT,
    )

    parser.add_argument(
        '--extend-starttype',
        help='Filter by service start type, '
        'appended to the default or the `--starttype` list. '
        'Can be specified multiple times. '
        'Example: `--extend-starttype=manual`.',
        dest='EXTEND_STARTTYPE',
        action='append',
        choices=STARTTYPES,
        default=None,
    )

    parser.add_argument(
        '--extend-status',
        help='Expected service status, '
        'appended to the default or the `--status` list. '
        'Can be specified multiple times. '
        'Example: `--extend-status=start_pending`.',
        dest='EXTEND_STATUS',
        action='append',
        choices=STATUSES,
        default=None,
    )

    parser.add_argument(
        '--service',
        help='Name of the Windows service(s) to check. '
        'Uses Python regular expressions. Case-insensitive.',
        dest='SERVICE',
        required=True,
    )

    parser.add_argument(
        '--starttype',
        help='Filter by service start type. '
        'Replaces the default list, use `--extend-starttype` to append to it. '
        'Can be specified multiple times. '
        f'Default: {", ".join(DEFAULT_STARTTYPE)}.',
        dest='STARTTYPE',
        action='append',
        choices=STARTTYPES,
    )

    parser.add_argument(
        '--status',
        help='Expected service status. '
        'Replaces the default list, use `--extend-status` to append to it. '
        'Can be specified multiple times. '
        f'Default: {", ".join(DEFAULT_STATUS)}.',
        dest='STATUS',
        default=None,  # due to https://bugs.python.org/issue16399, see in main() below
        action='append',
        choices=STATUSES,
    )

    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 number of services in the expected status. '
        'Supports Nagios ranges. '
        'Default: %(default)s',
        dest='WARN',
        default=DEFAULT_WARN,
    )

    args, _ = parser.parse_known_args()
    return args


def collect_matches(compiled_regex, test_arg):
    """Return every Windows service whose name matches `compiled_regex`.

    Walks `psutil.win_service_iter()` in production and the
    `--test` fixture (CSV lines of `name, display_name, status,
    start_type`) in unit-test mode. Each match is a flat dict, so
    the caller can apply the start-type and status filters without
    having to re-wrap anything. A `psutil.NoSuchProcess` raised
    mid-iteration (race condition: a service disappears between
    enumeration and inspection) is swallowed so the caller still
    gets whatever was already collected.
    """
    matches = []
    try:
        if test_arg is None:
            for s in psutil.win_service_iter():
                if re.search(compiled_regex, s.name()):
                    matches.append(
                        {
                            'name': s.name(),
                            'display_name': s.display_name(),
                            'status': s.status(),
                            'start_type': s.start_type(),
                        }
                    )
            return matches
        stdout, _, _ = lib.lftest.test(test_arg)
        for line in stdout.splitlines():
            name, display_name, status, start_type = line.split(', ')
            if re.search(compiled_regex, name):
                matches.append(
                    {
                        'name': name,
                        'display_name': display_name,
                        'status': status,
                        'start_type': start_type,
                    }
                )
    except psutil.NoSuchProcess:
        pass
    except (PermissionError, psutil.AccessDenied):
        # Windows lets services, interactive sessions and administrators list the
        # services, but not a standard user who logged in remotely: the service
        # control manager grants SC_MANAGER_ENUMERATE_SERVICE to IU, SU, SY and BA
        # only. Measured on Windows Server 2019 over SSH.
        lib.base.cu(
            'Access denied while listing the Windows services. The account the'
            ' check runs as may not list them from a remote login such as SSH or'
            ' WinRM; the Icinga 2 agent, an interactive session and an'
            ' administrator may.',
            traceback=False,
        )
    return matches


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)

    if args.TEST is None and not lib.base.WINDOWS:
        lib.base.cu('This check runs on Windows only.')

    # set default values for append parameters that were not specified
    if args.EXTEND_STARTTYPE is None:
        args.EXTEND_STARTTYPE = []
    if args.EXTEND_STATUS is None:
        args.EXTEND_STATUS = []
    if args.STARTTYPE is None:
        args.STARTTYPE = DEFAULT_STARTTYPE
    # build a new list, `+=` would modify DEFAULT_STARTTYPE in place
    args.STARTTYPE = args.STARTTYPE + args.EXTEND_STARTTYPE
    if args.STATUS is None:
        args.STATUS = DEFAULT_STATUS
    # build a new list, `+=` would modify DEFAULT_STATUS in place
    args.STATUS = args.STATUS + args.EXTEND_STATUS

    # fetch data
    # regex-match first, start_type-filter second: otherwise a service
    # that would match the user's regex but happens to have a
    # non-matching start_type is silently dropped before the regex even
    # runs, and the user sees a "does not match any service name" error
    # for a service that very much exists. By splitting the two phases
    # we can give an actionable error when the regex found candidates
    # but the start_type filter removed all of them ([#921]).
    compiled_service_regex = lib.base.coe(
        lib.txt.compile_regex(args.SERVICE, '--service', flags=re.IGNORECASE)
    )

    regex_matches = collect_matches(compiled_service_regex, args.TEST)

    # no regex match at all — the name truly does not exist
    if not regex_matches:
        lib.base.cu(f'r`{args.SERVICE}` does not match any service name.')

    # init some vars
    msg = ''
    state = STATE_OK
    table_data = []
    svcstates = {
        'continue_pending': 0,
        'pause_pending': 0,
        'paused': 0,
        'running': 0,
        'start_pending': 0,
        'stop_pending': 0,
        'stopped': 0,
    }
    svcstate_cnt = 0  # this is the overall count of matching service states

    # analyze data
    # regex matched, now apply the start_type filter
    for match in regex_matches:
        if match['start_type'] not in args.STARTTYPE:
            continue
        if match['status'] in args.STATUS:
            svcstates[match['status']] += 1
            svcstate_cnt += 1
        table_data.append(match)

    # regex found candidates but all were rejected by --starttype: tell
    # the user exactly which start_types we saw so they can fix the
    # filter instead of chasing a phantom "does not match" error.
    if not table_data:
        seen_types = sorted({m['start_type'] for m in regex_matches})
        sample = ', '.join(
            f'{m["name"]} ({m["start_type"]})' for m in regex_matches[:5]
        )
        if len(regex_matches) > 5:
            sample += ', ...'
        lib.base.cu(
            f'r`{args.SERVICE}` matched {len(regex_matches)} '
            f'{lib.txt.pluralize("service", len(regex_matches))}, but none '
            f'have a start_type in {sorted(args.STARTTYPE)}. Matched '
            f'start_types: {seen_types}. Matched services: {sample}. '
            f'Adjust --starttype to include them.'
        )
    svc_cnt = len(table_data)

    # build the message
    # alert: e.g. at least 10 but not more than 20 Windows-Services must meet any given --status
    state = lib.base.get_state(svcstate_cnt, args.WARN, args.CRIT, _operator='range')
    msg += lib.base.get_table(
        table_data,
        [
            'display_name',
            'name',
            'status',
            'start_type',
        ],
        header=[
            'Display Name',
            'Service Name',
            'Status',
            'Startup',
        ],
    )
    msg = (
        f'{"Everything is ok. " if state == STATE_OK else ""}'
        f'{svc_cnt} '
        f'{lib.txt.pluralize("service", svc_cnt)} '
        f'named r`{args.SERVICE}` '
        f'and start type {args.STARTTYPE} found, '
        f'{svcstate_cnt} in status {args.STATUS} '
        f'(thresholds {args.WARN}/{args.CRIT})'
        f'{lib.base.state2str(state, prefix=" ")}.\n\n'
    ) + msg

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


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