#!/usr/bin/env python3
# -*- encoding: utf-8; py-indent-offset: 4 -*-
# Sophos XGS Certs
# +------------------------------------------------------------------+
#      _____ ____   _____    _____  ____  _               _____  
#     |_   _|  _ \ / ____|  / ____|/ __ \| |        /\   |  __ \ 
#       | | | |_) | |      | (___ | |  | | |       /  \  | |__) |
#       | | |  _ <| |       \___ \| |  | | |      / /\ \ |  _  / 
#      _| |_| |_) | |____   ____) | |__| | |____ / ____ \| | \ \ 
#     |_____|____/ \_____| |_____/ \____/|______/_/    \_\_|  \_\
#                                                                                                                        
# +------------------------------------------------------------------+
# Copyright  IBC SOLAR AG - License: GNU General Public License v2.
# Author: Marvin Sobeck
# Link: https://www.ibc-solar.com/

import argparse
import io
import json
import os
import re
import ssl
import subprocess
import sys
import tarfile
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
from pathlib import PurePosixPath
from xml.sax.saxutils import escape


MAXIMUM_ARCHIVE_SIZE = 104857600

PRIVATE_FILE_EXTENSIONS = (
    ".key",
    ".p12",
    ".pfx",
    ".pkcs12",
)

PUBLIC_CERTIFICATE_EXTENSIONS = (
    ".pem",
    ".crt",
    ".cer",
    ".cert",
    ".der",
)


def parse_args():
    parser = argparse.ArgumentParser(
        description="Checkmk special agent for Sophos XGS certificates"
    )

    parser.add_argument(
        "--url",
        required=True,
        help="Sophos API base URL",
    )

    parser.add_argument(
        "--username",
        required=True,
        help="Sophos API username",
    )

    parser.add_argument(
        "--password",
        required=True,
        help="Sophos API password",
    )

    parser.add_argument(
        "--timeout",
        type=int,
        default=30,
        help="HTTP timeout in seconds",
    )

    parser.add_argument(
        "--no-cert-check",
        action="store_true",
        help="Disable TLS certificate validation",
    )

    parser.add_argument(
        "--include-per-user",
        action="store_true",
        help="Include Sophos per-user certificates",
    )

    parser.add_argument(
        "--debug",
        action="store_true",
        help="Print debug information to stderr",
    )

    return parser.parse_args()


def debug(args, message):
    if args.debug:
        print(
            "DEBUG: %s" % message,
            file=sys.stderr,
        )


def local_name(tag):
    return str(tag).rsplit("}", 1)[-1]


def normalize_value(value):
    return re.sub(
        r"[^a-z0-9]+",
        "",
        str(value or "").lower(),
    )


def text_or_none(element):
    if element is None:
        return None

    if element.text is None:
        return None

    value = element.text.strip()

    if not value:
        return None

    return value


def flatten_element(element):
    data = {}

    for child in element.iter():
        if child is element:
            continue

        key = local_name(child.tag)
        value = text_or_none(child)

        if value is None:
            continue

        if key not in data:
            data[key] = value
            continue

        existing_value = str(data[key])

        if value not in existing_value:
            data[key] = "%s | %s" % (
                existing_value,
                value,
            )

    return data


def get_first(data, candidates):
    normalized_data = {
        normalize_value(key): value
        for key, value in data.items()
    }

    for candidate in candidates:
        value = normalized_data.get(
            normalize_value(candidate)
        )

        if value not in (None, ""):
            return value

    return None


def extract_cn(subject):
    if not subject:
        return None

    match = re.search(
        r"\bCN\s*=\s*([^,/]+)",
        str(subject),
        flags=re.IGNORECASE,
    )

    if match:
        return match.group(1).strip()

    return None


def parse_openssl_date(value):
    if "=" in value:
        value = value.split(
            "=",
            1,
        )[1].strip()

    for date_format in (
        "%b %d %H:%M:%S %Y %Z",
        "%b  %d %H:%M:%S %Y %Z",
    ):
        try:
            parsed_date = datetime.strptime(
                value,
                date_format,
            )

            return parsed_date.replace(
                tzinfo=timezone.utc
            ).strftime(
                "%Y-%m-%d %H:%M:%S"
            )

        except ValueError:
            continue

    return None


def run_openssl(
    certificate_data,
    certificate_format,
):
    command = [
        "openssl",
        "x509",
    ]

    if certificate_format == "DER":
        command.extend(
            [
                "-inform",
                "DER",
            ]
        )

    command.extend(
        [
            "-noout",
            "-subject",
            "-issuer",
            "-dates",
            "-serial",
            "-fingerprint",
            "-sha256",
        ]
    )

    return subprocess.run(
        command,
        input=certificate_data,
        capture_output=True,
        check=True,
        timeout=10,
    )


def read_certificate_with_openssl(
    certificate_data,
    archive_member_name,
):
    result = None
    detected_format = None

    for certificate_format in (
        "PEM",
        "DER",
    ):
        try:
            result = run_openssl(
                certificate_data,
                certificate_format,
            )

            detected_format = certificate_format
            break

        except Exception:
            continue

    if result is None:
        return None

    output = result.stdout.decode(
        "utf-8",
        errors="replace",
    )

    normalized_member_name = str(
        archive_member_name
    ).replace(
        "\\",
        "/",
    )

    member_path = PurePosixPath(
        normalized_member_name
    )

    certificate_information = {
        "subject": None,
        "issuer": None,
        "valid_from": None,
        "valid_until": None,
        "serial": None,
        "fingerprint_sha256": None,
        "cn": None,
        "file_name": member_path.name,
        "file_stem": member_path.stem,
        "format": detected_format,
    }

    for line in output.splitlines():
        line = line.strip()

        if line.startswith("subject="):
            certificate_information["subject"] = (
                line.split(
                    "=",
                    1,
                )[1].strip()
            )

            certificate_information["cn"] = extract_cn(
                certificate_information["subject"]
            )

        elif line.startswith("issuer="):
            certificate_information["issuer"] = (
                line.split(
                    "=",
                    1,
                )[1].strip()
            )

        elif line.startswith("notBefore="):
            certificate_information["valid_from"] = (
                parse_openssl_date(line)
            )

        elif line.startswith("notAfter="):
            certificate_information["valid_until"] = (
                parse_openssl_date(line)
            )

        elif line.startswith("serial="):
            certificate_information["serial"] = (
                line.split(
                    "=",
                    1,
                )[1].strip()
            )

        elif "Fingerprint=" in line:
            certificate_information["fingerprint_sha256"] = (
                line.split(
                    "=",
                    1,
                )[1].strip()
            )

    return certificate_information


def normalized_archive_name(member_name):
    return str(
        member_name
    ).replace(
        "\\",
        "/",
    ).lower()


def archive_base_name(member_name):
    normalized_name = normalized_archive_name(
        member_name
    )

    return PurePosixPath(
        normalized_name
    ).name


def is_private_member(member_name):
    normalized_name = normalized_archive_name(
        member_name
    )

    base_name = archive_base_name(
        member_name
    )

    if "privatekeyfile" in normalized_name:
        return True

    for extension in PRIVATE_FILE_EXTENSIONS:
        if base_name.endswith(extension):
            return True

    return False


def is_entities_member(member_name):
    return (
        archive_base_name(member_name)
        == "entities.xml"
    )


def is_public_certificate_member(
    member_name,
):
    normalized_name = normalized_archive_name(
        member_name
    )

    base_name = archive_base_name(
        member_name
    )

    if is_private_member(member_name):
        return False

    if "certificatefile" in normalized_name:
        return True

    for extension in PUBLIC_CERTIFICATE_EXTENSIONS:
        if base_name.endswith(extension):
            return True

    return False


def read_tar_member(archive, member):
    if not member.isfile():
        return None

    member_file = archive.extractfile(
        member
    )

    if member_file is None:
        return None

    return member_file.read()


def process_archive_in_memory(
    archive_data,
    args,
):
    entities_data = None
    certificate_index = []

    archive_buffer = io.BytesIO(
        archive_data
    )

    with tarfile.open(
        fileobj=archive_buffer,
        mode="r:",
    ) as archive:
        for member in archive.getmembers():
            member_name = member.name

            if not member.isfile():
                continue

            if is_private_member(member_name):
                debug(
                    args,
                    "Skipping private-key archive member: %s"
                    % member_name,
                )
                continue

            if is_entities_member(member_name):
                member_data = read_tar_member(
                    archive,
                    member,
                )

                if member_data is None:
                    continue

                if entities_data is not None:
                    raise RuntimeError(
                        "More than one Entities.xml found "
                        "in Sophos Certificate archive"
                    )

                entities_data = member_data

                debug(
                    args,
                    "Loaded Entities.xml into memory: %s"
                    % member_name,
                )
                continue

            if not is_public_certificate_member(
                member_name
            ):
                debug(
                    args,
                    "Skipping unrelated archive member: %s"
                    % member_name,
                )
                continue

            member_data = read_tar_member(
                archive,
                member,
            )

            if member_data is None:
                continue

            parsed_certificate = (
                read_certificate_with_openssl(
                    member_data,
                    member_name,
                )
            )

            if parsed_certificate is None:
                debug(
                    args,
                    "Could not parse public certificate: %s"
                    % member_name,
                )
                continue

            certificate_index.append(
                parsed_certificate
            )

            debug(
                args,
                (
                    "Parsed public certificate from memory: "
                    "file=%s, cn=%s, valid_from=%s, "
                    "valid_until=%s, format=%s"
                )
                % (
                    parsed_certificate.get(
                        "file_name"
                    ),
                    parsed_certificate.get(
                        "cn"
                    ),
                    parsed_certificate.get(
                        "valid_from"
                    ),
                    parsed_certificate.get(
                        "valid_until"
                    ),
                    parsed_certificate.get(
                        "format"
                    ),
                ),
            )

    if entities_data is None:
        raise RuntimeError(
            "Entities.xml not found in "
            "Sophos Certificate archive"
        )

    return (
        entities_data,
        certificate_index,
    )


def values_equal(value_a, value_b):
    normalized_a = normalize_value(
        value_a
    )

    normalized_b = normalize_value(
        value_b
    )

    return bool(
        normalized_a
        and normalized_b
        and normalized_a == normalized_b
    )


def find_unique_match(
    certificate_index,
    field_name,
    value,
):
    if not value:
        return None

    matches = []

    for certificate in certificate_index:
        if values_equal(
            certificate.get(field_name),
            value,
        ):
            matches.append(
                certificate
            )

    if len(matches) == 1:
        return matches[0]

    return None


def find_referenced_cert(
    entity_data,
    certificate_index,
):
    file_reference = get_first(
        entity_data,
        [
            "CertificateFile",
            "CertFile",
            "FileName",
        ],
    )

    if file_reference:
        normalized_file_reference = str(
            file_reference
        ).replace(
            "\\",
            "/",
        )

        file_reference_path = PurePosixPath(
            normalized_file_reference
        )

        certificate = find_unique_match(
            certificate_index,
            "file_name",
            file_reference_path.name,
        )

        if certificate:
            return certificate

        certificate = find_unique_match(
            certificate_index,
            "file_stem",
            file_reference_path.stem,
        )

        if certificate:
            return certificate

    entity_serial = get_first(
        entity_data,
        [
            "Serial",
            "SerialNumber",
            "CertificateSerial",
        ],
    )

    certificate = find_unique_match(
        certificate_index,
        "serial",
        entity_serial,
    )

    if certificate:
        return certificate

    subject = get_first(
        entity_data,
        [
            "Subject",
            "SubjectName",
            "SubjectDN",
        ],
    )

    certificate = find_unique_match(
        certificate_index,
        "subject",
        subject,
    )

    if certificate:
        return certificate

    entity_cn = get_first(
        entity_data,
        [
            "CommonName",
            "CN",
        ],
    )

    if not entity_cn:
        entity_cn = extract_cn(
            subject
        )

    certificate = find_unique_match(
        certificate_index,
        "cn",
        entity_cn,
    )

    if certificate:
        return certificate

    name = get_first(
        entity_data,
        [
            "Name",
            "CertificateName",
            "CertName",
            "DisplayName",
        ],
    )

    certificate = find_unique_match(
        certificate_index,
        "cn",
        name,
    )

    if certificate:
        return certificate

    return None


def load_certificate_elements(
    entities_data,
    args,
):
    try:
        root = ET.fromstring(
            entities_data
        )

    except ET.ParseError as exception:
        raise RuntimeError(
            "Could not parse Entities.xml: %s"
            % exception
        )

    elements = []

    for element in root.iter():
        element_name = normalize_value(
            local_name(
                element.tag
            )
        )

        if element_name == "certificate":
            elements.append(
                element
            )

    if not elements:
        raise RuntimeError(
            "No Certificate entries found "
            "in Entities.xml"
        )

    debug(
        args,
        "Loaded %d Certificate entries"
        % len(elements),
    )

    return elements


def parse_certificate_entities(
    entities_data,
    certificate_index,
    args,
):
    certificates = []

    elements = load_certificate_elements(
        entities_data,
        args,
    )

    for element in elements:
        data = flatten_element(
            element
        )

        name = get_first(
            data,
            [
                "Name",
                "CertificateName",
                "CertName",
            ],
        )

        if not name:
            continue

        action = get_first(
            data,
            [
                "Action",
            ],
        )

        normalized_action = str(
            action or ""
        ).strip().lower()

        if (
            normalized_action == "p"
            and not args.include_per_user
        ):
            debug(
                args,
                (
                    "Ignoring per-user certificate: "
                    "name=%s, action=%s"
                )
                % (
                    name,
                    action,
                ),
            )
            continue

        certificate_file = get_first(
            data,
            [
                "CertificateFile",
                "CertFile",
                "FileName",
            ],
        )

        certificate_format = get_first(
            data,
            [
                "CertificateFormat",
                "Format",
            ],
        )

        trusted = get_first(
            data,
            [
                "Trusted",
                "IsTrusted",
            ],
        )

        subject = get_first(
            data,
            [
                "Subject",
                "SubjectName",
                "SubjectDN",
            ],
        )

        issuer = get_first(
            data,
            [
                "Issuer",
                "IssuerName",
                "IssuedBy",
            ],
        )

        valid_from = get_first(
            data,
            [
                "ValidFrom",
                "NotBefore",
            ],
        )

        valid_until = get_first(
            data,
            [
                "ValidUntil",
                "NotAfter",
                "ExpiryDate",
                "Expiry",
            ],
        )

        referenced_certificate = (
            find_referenced_cert(
                data,
                certificate_index,
            )
        )

        if referenced_certificate:
            debug(
                args,
                (
                    "Matched certificate '%s' "
                    "to public certificate '%s' "
                    "(valid_until=%s)"
                )
                % (
                    name,
                    referenced_certificate.get(
                        "file_name"
                    ),
                    referenced_certificate.get(
                        "valid_until"
                    ),
                ),
            )

            subject = (
                subject
                or referenced_certificate.get(
                    "subject"
                )
            )

            issuer = (
                issuer
                or referenced_certificate.get(
                    "issuer"
                )
            )

            valid_from = (
                valid_from
                or referenced_certificate.get(
                    "valid_from"
                )
            )

            valid_until = (
                valid_until
                or referenced_certificate.get(
                    "valid_until"
                )
            )

        else:
            debug(
                args,
                (
                    "No public certificate match: "
                    "name=%s, action=%s, file=%s"
                )
                % (
                    name,
                    action,
                    certificate_file,
                ),
            )

        item = {
            "name": name,
            "action": action,
            "certificate_file": certificate_file,
            "format": certificate_format,
            "trusted": trusted,
            "subject": subject,
            "issuer": issuer,
            "valid_from": valid_from,
            "valid_until": valid_until,
        }

        if referenced_certificate:
            item["cn"] = (
                referenced_certificate.get(
                    "cn"
                )
            )

            item["serial"] = (
                referenced_certificate.get(
                    "serial"
                )
            )

            item["fingerprint_sha256"] = (
                referenced_certificate.get(
                    "fingerprint_sha256"
                )
            )

        certificates.append(
            item
        )

    return certificates


def build_request_xml(args):
    return (
        '<Request APIVersion="2200.1">'
        "<Login>"
        "<Username>%s</Username>"
        "<Password>%s</Password>"
        "</Login>"
        "<Get><Certificate/></Get>"
        "</Request>"
    ) % (
        escape(
            args.username
        ),
        escape(
            args.password
        ),
    )


def fetch_certificate_archive(
    args,
):
    request_xml = build_request_xml(
        args
    )

    query = urllib.parse.urlencode(
        {
            "reqxml": request_xml,
        }
    )

    url = (
        "%s/webconsole/APIController?%s"
        % (
            args.url.rstrip("/"),
            query,
        )
    )

    context = None

    if args.no_cert_check:
        context = (
            ssl._create_unverified_context()
        )

    request = urllib.request.Request(
        url,
        headers={
            "User-Agent": (
                "checkmk-sophos-xgs-certs"
            ),
            "Accept": "*/*",
        },
    )

    with urllib.request.urlopen(
        request,
        timeout=args.timeout,
        context=context,
    ) as response:
        response_data = response.read(
            MAXIMUM_ARCHIVE_SIZE + 1
        )

        content_type = (
            response.headers.get(
                "Content-Type",
                "",
            )
        )

    if not response_data:
        raise RuntimeError(
            "Sophos API returned "
            "an empty response"
        )

    if (
        len(response_data)
        > MAXIMUM_ARCHIVE_SIZE
    ):
        raise RuntimeError(
            "Sophos Certificate archive "
            "exceeds the configured size limit"
        )

    if (
        response_data.lstrip().startswith(
            b"<?xml"
        )
        or "text/xml"
        in content_type.lower()
    ):
        raise RuntimeError(
            "Sophos API error: %s"
            % response_data.decode(
                "utf-8",
                errors="replace",
            )
        )

    debug(
        args,
        (
            "Downloaded Sophos Certificate archive "
            "into memory: %d bytes"
        )
        % len(response_data),
    )

    return response_data


def print_section(name, payload):
    print(
        "<<<%s>>>" % name
    )

    print(
        json.dumps(
            payload,
            sort_keys=True,
            default=str,
            ensure_ascii=False,
            separators=(",", ":"),
        )
    )


def main():
    args = parse_args()

    try:
        archive_data = (
            fetch_certificate_archive(
                args
            )
        )

        (
            entities_data,
            certificate_index,
        ) = process_archive_in_memory(
            archive_data,
            args,
        )

        certificates = (
            parse_certificate_entities(
                entities_data,
                certificate_index,
                args,
            )
        )

        print_section(
            "sophos_xgs_certs",
            certificates,
        )

    except Exception as exception:
        print_section(
            "sophos_xgs_certs",
            [],
        )

        print_section(
            "sophos_xgs_certs_status",
            {
                "state": "CRIT",
                "message": str(
                    exception
                ),
            },
        )

        return 0

    return 0


if __name__ == "__main__":
    raise SystemExit(
        main()
    )