#!/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 Apache httpd server discloses about itself in its HTTP
responses: the product and version banner in the `Server` response header, the server
signature footer on generated error pages, and inode numbers leaked through `ETag` response
headers. Every finding is measured on the response the server actually sends, so a value
overridden further down the configuration or by a reverse 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,
its signature or an inode number."""

# The checks follow the "Information Leakage" chapter of the CIS Apache HTTP
# Server 2.4 Benchmark: the `Server` header, the server signature footer and
# inode numbers in `ETag`. 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 signature footer we are looking for.
PROBE_PATH = '/linuxfabrik-monitoring-plugins-probe'

# Apache renders the signature as
# "<address>{banner} Server at {name} Port {port}</address>", for both
# "ServerSignature On" and "ServerSignature EMail" (the latter wraps the server
# name in a mailto link, which the non-greedy middle part absorbs).
# See ap_psignature() in server/core.c.
SIGNATURE_REGEX = re.compile(
    r'<address>(?P<banner>.*?) Server at .*? Port \d+</address>',
    re.IGNORECASE | re.DOTALL,
)

# Response headers that name the backend product or its version. The Apache
# benchmark has no control for these, verified against all five published
# revisions, but a proxied `X-Powered-By` discloses the application stack no
# matter which server forwarded it, so the check is carried out anyway.
BACKEND_HEADERS = (
    'x-aspnet-version',
    'x-aspnetmvc-version',
    'x-generator',
    'x-powered-by',
)

# An ETag built with the INode field has three hyphen-separated hexadecimal
# components (inode-size-mtime); without it, two (size-mtime). Weak validators
# ("W/") and the surrounding quotes are stripped before matching.
ETAG_INODE_REGEX = re.compile(r'^[0-9a-f]+-[0-9a-f]+-[0-9a-f]+$', re.IGNORECASE)


# Directive that suppresses the version banner. Only ever offered when the
# response really came from Apache httpd: 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 `ServerTokens Prod` to reduce the `Server` header to `Apache`.'

# 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 Apache httpd 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 check_server_banner(response, severity_state):
    """The `Server` header must not disclose a version.

    Only a bare product token ("Apache") is compliant. Everything the
    ServerTokens directive adds on top starts with a slash ("Apache/2.4"), 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 describe_tokens(banner):
    """Name what a `Server` header discloses beyond the product token."""
    if '(' in banner:
        # "Apache/2.4.62 (Rocky Linux)" and anything appended by a module.
        if banner.rstrip().endswith(')'):
            return 'version and operating system'
        return 'version, operating system and loaded modules'
    version = banner.partition('/')[2].strip()
    if version.count('.') >= 2:
        return 'the full version'
    return 'a partial version'


def find_signature(responses):
    """Return the first server signature found, with the page it was found on.

    Apache writes the signature onto every page it generates itself, which is
    an error page as well as a directory listing. Both 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 = SIGNATURE_REGEX.search(response['response'])
        if match:
            return (match.group('banner'), label, generated)
    return None


def check_server_signature(signature, severity_state):
    """Generated pages must not carry the signature footer."""
    if signature is None:
        return (
            STATE_OK,
            'no signature footer',
            'No page served here carries a server signature.',
            None,
        )
    banner, label, generated = signature
    if generated:
        recommendation = 'Set `ServerSignature Off`.'
    else:
        # Away from the error probe the page may be a directory listing the
        # server generated, or a file somebody stored that happens to contain a
        # captured error page. Both disclose the version, but only the first one
        # is fixed by the directive.
        recommendation = (
            'Set `ServerSignature Off` if the server generated this page. If it '
            'is stored content, remove the version from the file itself.'
        )
    return (
        severity_state,
        f'`{banner} Server at ...`',
        f'The page served at {label} carries a server signature footer.',
        recommendation,
    )


def check_etag_inode(response, severity_state):
    """`ETag` must not be built from the file's inode number."""
    if response is None:
        return (
            None,
            'not evaluated',
            'The URL did not answer, so no `ETag` could be read.',
            None,
        )
    etag = response['response_header'].get('etag', '')
    if not etag:
        return (
            STATE_OK,
            'no `ETag` response header',
            'Without an `ETag` there is no inode to disclose.',
            None,
        )
    # Weak validators are prefixed with W/ and the value itself is quoted.
    value = etag.strip()
    if value.upper().startswith('W/'):
        value = value[2:]
    value = value.strip('"')
    if not ETAG_INODE_REGEX.match(value):
        return (
            STATE_OK,
            f'`ETag: {etag}`',
            'Built from size and modification time only.',
            None,
        )
    return (
        severity_state,
        f'`ETag: {etag}`',
        f'Discloses the inode number `{value.split("-")[0]}`.',
        'Set `FileETag MTime Size` to drop the inode component.',
    )


# 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 server 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.',
        'Drop the header before it reaches the client with '
        '`Header always unset X-Powered-By` from `mod_headers`, and turn it off '
        f'at its 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 signature footer 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
    # The `Server` header and the signature come from the error probe, the
    # `ETag` from the base URL, because an error response carries no ETag.
    # A cache or reverse proxy in front of the server answers with its own
    # `Server` header and rewrites `ETag`, so neither says anything about the
    # origin any more. The signature footer survives, because it is part of the
    # page body the proxy passes through, and it is proof that an Apache
    # generated that page. Without that proof and with a foreign banner there is
    # no Apache to audit here, so the check says where to point instead.
    product = lib.url.server_product(error_response['response_header'])
    signature = find_signature(
        [
            ('the probed error page', error_response, True),
            ('the base URL', main_response, False),
        ]
    )
    fronted = product not in (None, 'apache')
    if fronted and signature is None:
        lib.base.oao(
            f'This URL is answered by {product}, not by Apache httpd. '
            f'Point `--url` at the Apache httpd 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),
        ),
        (
            'Server signature',
            check_server_signature(signature, severity_state),
        ),
        (
            'ETag inode',
            (None, 'not evaluated', FRONTED_DETAIL, None)
            if fronted
            else check_etag_inode(main_response, 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(
        'apache_httpd_disclosures',
        disclosed,
        uom=None,
        _min=0,
        _max=len(checks),
    )
    perfdata += lib.base.get_perfdata(
        'apache_httpd_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()
