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

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

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

DESCRIPTION = """Checks how much an NGINX server discloses about itself and about the
application behind it in its HTTP responses: the product and version banner in the `Server`
response header, references to the server software in generated error pages, and headers an
upstream application leaks through the proxy, such as `X-Powered-By`. Every finding is
measured on the response the server actually sends, so a value overridden further down the
configuration or by another proxy is reported as it reaches a client rather than as it is
written in a configuration file. Each finding maps to a copy-pasteable configuration
recommendation. Alerts when the server discloses its version, names itself on an error page,
or forwards a header that identifies the backend."""

# The checks follow the information disclosure controls of the CIS NGINX
# Benchmark (v3.0.0): the `server_tokens` directive, the product name in the
# default error pages, and headers an upstream leaks through the proxy. The
# benchmark numbering is deliberately not printed, so the output does not age
# with the document.

DEFAULT_INSECURE = False
DEFAULT_NO_PROXY = False
DEFAULT_SEVERITY = 'warn'
DEFAULT_TIMEOUT = 8
DEFAULT_URL = 'http://localhost'

# Path appended to --url to provoke a server-generated error page. Static and
# self-describing, so an admin who finds it in the access log can tell it apart
# from a scanner. A path that exists would be served normally and would never
# carry the default error page we are looking for.
PROBE_PATH = '/linuxfabrik-monitoring-plugins-probe'

# Response headers that name the backend product or its version. The benchmark
# calls out `X-Powered-By` and `Server`; `Server` has its own check here, and
# the remaining entries are the same idea under the names other stacks use.
BACKEND_HEADERS = (
    'x-aspnet-version',
    'x-aspnetmvc-version',
    'x-generator',
    'x-powered-by',
)

# NGINX names itself in the footer of every default error page, and keeps doing
# so with `server_tokens off`, which only removes the version from that same
# footer. See ngx_http_error_tail in src/http/ngx_http_special_response.c.
# The benchmark searches the whole body for the bare product name, which reports
# any page that merely mentions NGINX, a directory listing of a package named
# `nginx-mode` for example. Matching the generated footer instead keeps the
# finding to pages the server really produced.
PRODUCT_FOOTER_REGEX = re.compile(
    r'<hr>\s*<center>\s*(?P<footer>nginx[^<]*)</center>',
    re.IGNORECASE,
)

# Directive that suppresses the version banner. Only ever offered when the
# response really came from NGINX: recommending a directive of a product the
# host does not run is worse than saying nothing, so a foreign server is named
# instead and the operator is pointed at the right endpoint.
VERSION_KNOB = 'Set `server_tokens off;` to reduce the `Server` header to `nginx`.'

# Shown for the checks a cache or reverse proxy in front of the server makes
# unanswerable, because it replaces the header they read.
FRONTED_DETAIL = (
    'A cache or proxy in front of the server replaces this response header, so '
    'it says nothing about the origin.'
)


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(
        '--insecure',
        help=lib.args.help('--insecure'),
        dest='INSECURE',
        action='store_true',
        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=lib.args.help('--no-proxy'),
        dest='NO_PROXY',
        action='store_true',
        default=DEFAULT_NO_PROXY,
    )

    parser.add_argument(
        '--proxy',
        help=lib.args.help('--proxy'),
        dest='PROXY',
        default=None,
    )

    parser.add_argument(
        '--severity',
        help='State to report for a server that discloses information. '
        'One of `warn` or `crit`. '
        'Default: %(default)s',
        dest='SEVERITY',
        choices=['warn', 'crit'],
        default=DEFAULT_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,
    )

    parser.add_argument(
        '--url',
        help='Base URL of the NGINX server to inspect. '
        'Example: `--url=https://www.example.com` '
        'Default: %(default)s',
        dest='URL',
        default=DEFAULT_URL,
    )

    args, _ = parser.parse_known_args()
    return args


def fetch_response(url, args):
    """Fetch one URL and return the extended response dict.

    A 4xx or 5xx counts as success here: an error page is the primary subject
    of this check, because its header and its generated body carry exactly the
    information we are looking for. `lib.url.fetch()` reports such a status as
    a failure and hands the response dict back through `response_on_error`, so
    only a non-dict result is a real transport failure (connection refused,
    timeout, TLS error).
    """
    success, result = lib.url.fetch(
        url,
        extended=True,
        insecure=args.INSECURE,
        no_proxy=args.NO_PROXY,
        proxy=args.PROXY,
        response_on_error=True,
        timeout=args.TIMEOUT,
    )
    if not success and not isinstance(result, dict):
        return (False, result)
    return (True, result)


def describe_tokens(banner):
    """Name what a `Server` header discloses beyond the product token."""
    if '(' in banner:
        # "nginx/1.18.0 (Ubuntu)" and anything a build appends.
        if banner.rstrip().endswith(')'):
            return 'version and operating system'
        return 'version, operating system and build details'
    version = banner.partition('/')[2].strip()
    if version.count('.') >= 2:
        return 'the full version'
    return 'a partial version'


def check_server_banner(response, severity_state):
    """The `Server` header must not disclose a version.

    `server_tokens off` reduces the header to the bare product token, which is
    what the benchmark asks for. Everything the directive leaves in place starts
    with a slash ("nginx/1.28.0"), so the presence of a version part is the
    whole test.
    """
    banner = response['response_header'].get('server', '')
    if not banner:
        return (
            STATE_OK,
            'no `Server` response header',
            'The server does not identify itself at all.',
            None,
        )
    if '/' not in banner:
        return (
            STATE_OK,
            f'`Server: {banner}`',
            'Product token only, no version disclosed.',
            None,
        )
    return (
        severity_state,
        f'`Server: {banner}`',
        f'Discloses {describe_tokens(banner)}.',
        VERSION_KNOB,
    )


def find_product_footer(responses):
    """Return the first generated footer found, with the page it was found on.

    Both the probed error page and the base URL are searched, because a cache or
    a proxy in front of the server frequently answers the error probe on its own
    while still passing a generated page through untouched.
    """
    for label, response, generated in responses:
        if response is None:
            continue
        match = PRODUCT_FOOTER_REGEX.search(response['response'])
        if match:
            return (match.group('footer').strip(), label, generated)
    return None


def check_error_page(footer, severity_state):
    """A generated page must not name the server software.

    The default footer survives `server_tokens off`, which only drops the
    version from it, so this is a separate finding from the banner and is fixed
    by pointing `error_page` at content of your own.
    """
    if footer is None:
        return (
            STATE_OK,
            'no product name in the generated pages',
            'No page served here names the server software.',
            None,
        )
    name, label, generated = footer
    recommendation = (
        'Point `error_page` at pages of your own that carry no product name, '
        'for example `error_page 404 /404.html;`.'
    )
    if not generated:
        # Away from the error probe the page may be one the server generated, or
        # a file somebody stored that happens to contain a captured error page.
        # Both disclose the product, but only the first one is fixed by the
        # directive.
        recommendation += (
            ' If it is stored content, remove the product name from the file itself.'
        )
    return (
        severity_state,
        f'`{name}` on {label}',
        'The default page identifies the server as NGINX even when the '
        '`Server` header does not.',
        recommendation,
    )


# How to stop a backend from announcing itself in the first place, keyed by a
# lower-cased fragment of the header value. Only a value we recognise gets a
# concrete instruction; anything else is pointed at "the application that sets
# it", because naming the wrong stack's setting is worse than naming none.
BACKEND_SOURCE_HINTS = (
    ('asp.net', 'removing the `X-Powered-By` custom header from the site'),
    ('drupal', 'the setting of the content management system'),
    ('express', "`app.disable('x-powered-by')` in the application"),
    ('php', '`expose_php = Off` in php.ini'),
)


def backend_source_hint(header):
    """Name where a disclosed backend header is set, as far as it is knowable.

    `X-Powered-By` is consulted first because it is the header the advice names;
    only if it is absent or unrecognised do the remaining ones get a say.
    """
    ordered = [header['x-powered-by']] if 'x-powered-by' in header else []
    ordered += [v for k, v in sorted(header.items()) if k != 'x-powered-by']
    for value in ordered:
        for fragment, hint in BACKEND_SOURCE_HINTS:
            if fragment in value.lower():
                return hint
    return 'the application that sets it'


def check_backend_headers(responses, severity_state):
    """The proxy must not forward headers naming the backend.

    Product-agnostic on purpose: an upstream `X-Powered-By` says the same thing
    whichever server passed it on. Both responses are searched, because a header
    the application sets on an error page is frequently absent from a cached or
    statically served page, and the other way round.
    """
    disclosed = {}
    answered = False
    for response in responses:
        if response is None:
            continue
        answered = True
        for header in BACKEND_HEADERS:
            value = response['response_header'].get(header, '').strip()
            # A header set to a placeholder such as "-" is present but names
            # nothing, so it discloses nothing either.
            if header not in disclosed and any(c.isalnum() for c in value):
                disclosed[header] = value
    if not answered:
        return (
            None,
            'not evaluated',
            'The URL did not answer, so no response header could be read.',
            None,
        )
    if not disclosed:
        return (
            STATE_OK,
            'no backend headers',
            'Nothing identifies the application behind the server.',
            None,
        )
    found = [f'`{name}: {value}`' for name, value in disclosed.items()]
    return (
        severity_state,
        ', '.join(found),
        'The response names the backend technology and its version.',
        'Strip the header before it reaches the client with '
        '`proxy_hide_header X-Powered-By;`, or `fastcgi_hide_header X-Powered-By;` '
        'when the upstream is reached over FastCGI, and turn it off at its '
        f'source, {backend_source_hint(disclosed)}.',
    )


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
    # The error probe comes first and is the primary subject: it answers on any
    # server, needs no authentication, and its generated body is the only place
    # the default error page can appear.
    base_url = args.URL.rstrip('/')
    if args.TEST is None:
        error_response = lib.base.coe(
            fetch_response(base_url + PROBE_PATH, args),
        )
        success, main_response = fetch_response(base_url, args)
        if not success:
            main_response = None
    else:
        base = args.TEST[0]
        error_response = lib.lftest.test_http_response(args.TEST, f'{base}-error')
        if error_response is None:
            lib.base.cu(f'Fixture "{base}-error-header" not found.')
        main_response = lib.lftest.test_http_response(args.TEST, f'{base}-main')

    # init some vars
    state = STATE_OK
    perfdata = ''
    sections = []
    recommendations = []
    table_data = []
    severity_state = lib.base.str2state(args.SEVERITY)

    # analyze data
    # A cache or reverse proxy in front of the server answers with its own
    # `Server` header, so it says nothing about the origin any more. The
    # generated footer survives, because it is part of the page body the proxy
    # passes through, and it is proof that an NGINX produced that page. Without
    # that proof and with a foreign banner there is no NGINX to audit here, so
    # the check says where to point instead.
    product = lib.url.server_product(error_response['response_header'])
    footer = find_product_footer(
        [
            ('the probed error page', error_response, True),
            ('the base URL', main_response, False),
        ]
    )
    fronted = product not in (None, 'nginx')
    if fronted and footer is None:
        lib.base.oao(
            f'This URL is answered by {product}, not by NGINX. '
            f'Point `--url` at the NGINX instance you want to check.',
            STATE_UNKNOWN,
        )

    checks = [
        (
            'Server header',
            (None, f'not evaluated ({product})', FRONTED_DETAIL, None)
            if fronted
            else check_server_banner(error_response, severity_state),
        ),
        (
            'Error page',
            check_error_page(footer, severity_state),
        ),
        (
            'Backend headers',
            check_backend_headers([error_response, main_response], severity_state),
        ),
    ]

    disclosed = 0
    evaluated = 0
    for title, (item_state, result, detail, recommendation) in checks:
        # `item_state` is None for a check that could not be carried out. Such a
        # check must neither drive the overall state nor be counted as a pass.
        if item_state is not None:
            evaluated += 1
            state = lib.base.get_worst(state, item_state)
            if item_state != STATE_OK:
                disclosed += 1
        if recommendation:
            recommendations.append(recommendation)
        row = {
            'title': title,
            'result': result,
            'state': (
                ''
                if item_state is None
                else lib.base.state2str(item_state, empty_ok=False)
            ),
        }
        row['detail'] = detail
        table_data.append(row)

    # build the message
    skipped = len(checks) - evaluated
    if disclosed:
        summary = f'{disclosed} of {evaluated} checks report disclosed information.'
    else:
        summary = f'Everything is ok. Nothing disclosed in {evaluated} checks.'
    if skipped:
        summary += f' {skipped} not evaluated.'
    sections.append(summary)

    if recommendations:
        sections.append(
            'Recommendations:\n' + '\n'.join(f'* {r}' for r in recommendations)
        )

    perfdata += lib.base.get_perfdata(
        'nginx_disclosures',
        disclosed,
        uom=None,
        _min=0,
        _max=len(checks),
    )
    perfdata += lib.base.get_perfdata(
        'nginx_checks_evaluated',
        evaluated,
        uom=None,
        _min=0,
        _max=len(checks),
    )

    # build table output
    keys = ['title', 'result', 'detail', 'state']
    headers = ['Check', 'Result', 'Detail', 'State']
    sections.append(lib.base.get_table(table_data, keys, header=headers))

    msg = '\n\n'.join(sections)

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