#!/usr/bin/env python3
# Apprise
#
# SPDX-License-Identifier: GPL-2.0-only
# Line 2 above is the title Checkmk shows in the notification method dropdown;
# line 3 must stay free of "Bulk: yes" (bulk notifications are out of scope).
"""Checkmk notification script ``apprise``: forwards a notification to an Apprise API server.

Pipeline (each step is a small pure function, see below):
    env -> parse_config / parse_event -> map_type -> build_title / build_body
        -> build_payload -> send (bounded HTTP) -> classify_status -> exit code

Exit codes follow Checkmk: 0 sent, 1 temporary failure (retry later),
2 permanent failure (retry does not make sense).

Secrets/URLs are never printed: diagnostics only name the failing category.
"""

import base64
import html
import json
import os
import re
import socket
import ssl
import sys
import threading
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass, field

EXIT_OK = 0
EXIT_TEMPORARY = 1
EXIT_PERMANENT = 2

USER_AGENT = "checkmk-apprise-notification/1.0"
SUPPORTED_FORMATS = ("text", "html")  # no Markdown: Apprise cannot convert it to text targets
CONFIG_ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
TIMEOUT_RANGE = (1, 120)
DEFAULT_TIMEOUT = 10

# Upper bounds for values taken from Checkmk (worst-case body is roughly 6 KiB; the Apprise
# API accepts request bodies up to 3 MiB by default). Targets with smaller message limits
# are handled by Apprise itself (overflow setting of the target URL).
MAX_ID_CHARS = 255  # host, service, address, site, state, type, author
MAX_COMMENT_CHARS = 1000
MAX_OUTPUT_CHARS = 1500
MAX_LONG_OUTPUT_CHARS = 2000
MAX_TAG_CHARS = 200
DRAIN_BYTES = 4096  # response bodies are read (bounded) but never logged

PARAMETER_PREFIX = "NOTIFY_PARAMETER_"


class ConfigError(Exception):
    """Invalid local configuration or Checkmk event; retrying cannot help."""


@dataclass(frozen=True)
class Config:
    base_url: str
    config_id: str
    tag: str
    message_format: str
    verify_tls: bool
    timeout: int
    username: str = ""
    password: str = field(default="", repr=False)  # never print or log
    ca_file: str = ""


@dataclass(frozen=True)
class Event:
    what: str  # HOST | SERVICE
    notification_type: str  # PROBLEM, RECOVERY, ACKNOWLEDGEMENT, ...
    host: str
    address: str
    site: str
    state: str
    output: str
    service: str
    long_output: str
    author: str
    comment: str


@dataclass(frozen=True)
class Result:
    exit_code: int
    message: str


# --- parsing -------------------------------------------------------------------------------


def parse_bool(value: str, name: str) -> bool:
    lowered = value.strip().lower()
    if lowered in ("true", "1", "yes", "on"):
        return True
    if lowered in ("false", "0", "no", "off"):
        return False
    raise ConfigError(f"invalid boolean for {name}")


def validate_base_url(raw: str) -> str:
    parts = urllib.parse.urlsplit(raw.strip())
    if parts.scheme not in ("http", "https") or not parts.hostname:
        raise ConfigError("BASE_URL must be an http(s) URL")
    if parts.username is not None or parts.password is not None or "@" in parts.netloc:
        raise ConfigError("BASE_URL must not contain credentials")
    if parts.query or parts.fragment:
        raise ConfigError("BASE_URL must not contain a query or fragment")
    try:
        parts.port  # noqa: B018 - raises ValueError for an invalid port
    except ValueError:
        raise ConfigError("BASE_URL has an invalid port") from None
    return urllib.parse.urlunsplit((parts.scheme, parts.netloc, parts.path.rstrip("/"), "", ""))


def resolve_password(env: dict[str, str]) -> str:
    """Return the password from the Checkmk FormSpec ``Password`` parameter.

    Checkmk flattens ``("cmk_postprocessed", kind, (id, value))`` into
    NOTIFY_PARAMETER_PASSWORD_1 / _2 / _3_1 / _3_2. ``explicit_password`` carries the
    value itself; ``stored_password`` carries the password-store id, which only
    Checkmk's ``cmk.utils.password_store.extract`` can resolve (the same call
    Checkmk's bundled notification plugins use). It is imported lazily so that the
    script has no Checkmk dependency unless the password store is actually used.
    """
    prefix = PARAMETER_PREFIX + "PASSWORD_"
    kind = env.get(prefix + "2", "")
    if kind == "explicit_password":
        return env.get(prefix + "3_2", "")
    if kind == "stored_password":
        store_id = env.get(prefix + "3_1", "")
        if not store_id:
            raise ConfigError("password store entry is missing")
        try:
            from cmk.utils.password_store import extract  # noqa: PLC0415

            return extract(store_id)
        except Exception:  # noqa: BLE001 - never echo lower-level text
            raise ConfigError("password could not be read from the password store") from None
    return ""


MAX_CA_FILE_BYTES = 1024 * 1024


def site_root(env: dict[str, str]) -> str:
    """The Checkmk site directory, as exported to notification scripts."""
    return env.get("NOTIFY_OMD_ROOT") or env.get("OMD_ROOT") or ""


def validate_ca_file(raw: str, root: str) -> str:
    """Return the real path of a PEM file that holds the trusted CA/server certificate.

    The file must lie inside the Checkmk site directory (symbolic links are resolved first):
    the plug-in never reads files outside the site, as Checkmk Exchange extensions must not.
    """
    if not raw:
        return ""
    if len(raw) > 4096 or any(ch < " " or ch == "\x7f" for ch in raw):
        raise ConfigError("CA_FILE is not a valid path")
    path = os.path.expanduser(raw)
    if not os.path.isabs(path):
        raise ConfigError("CA_FILE must be an absolute path or start with ~/")
    if not root:
        raise ConfigError("CA_FILE needs the Checkmk site environment (OMD_ROOT)")
    path, root = os.path.realpath(path), os.path.realpath(root)
    if os.path.commonpath([path, root]) != root:
        raise ConfigError("CA_FILE must be inside the Checkmk site directory")
    if not os.path.isfile(path):
        raise ConfigError("CA_FILE does not exist or is not a regular file")
    try:
        if os.path.getsize(path) > MAX_CA_FILE_BYTES:
            raise ConfigError("CA_FILE is larger than 1 MiB")
        ssl.create_default_context(cafile=path)  # must contain at least one valid PEM certificate
    except PermissionError:
        raise ConfigError("CA_FILE cannot be read by the site user (check permissions)") from None
    except ssl.SSLError as error:  # before OSError: SSLError is a subclass of it
        reason = "".join(ch for ch in str(error.reason or "unknown") if ch.isprintable())[:60]
        raise ConfigError(f"CA_FILE is not a valid PEM certificate file ({reason})") from None
    except OSError:
        raise ConfigError("CA_FILE cannot be read") from None
    return path


def parse_config(env: dict[str, str]) -> Config:
    def param(name: str) -> str:
        return env.get(PARAMETER_PREFIX + name, "").strip()

    missing = [name for name in ("BASE_URL", "CONFIG_ID") if not param(name)]
    if missing:
        raise ConfigError("missing " + ", ".join(missing))

    config_id = param("CONFIG_ID")
    if not CONFIG_ID_RE.match(config_id):
        raise ConfigError("CONFIG_ID has an invalid format")

    message_format = param("MESSAGE_FORMAT") or "text"
    if message_format not in SUPPORTED_FORMATS:
        raise ConfigError("MESSAGE_FORMAT is not supported")

    verify_raw = param("VERIFY_TLS")
    verify_tls = parse_bool(verify_raw, "VERIFY_TLS") if verify_raw else True

    timeout_raw = param("TIMEOUT")
    try:
        timeout = int(timeout_raw) if timeout_raw else DEFAULT_TIMEOUT
    except ValueError:
        raise ConfigError("TIMEOUT must be an integer") from None
    if not TIMEOUT_RANGE[0] <= timeout <= TIMEOUT_RANGE[1]:
        raise ConfigError(f"TIMEOUT must be between {TIMEOUT_RANGE[0]} and {TIMEOUT_RANGE[1]}")

    tag = param("TAG")
    if len(tag) > MAX_TAG_CHARS or any(ch < " " or ch == "\x7f" for ch in tag):
        raise ConfigError("TAG is too long or contains control characters")

    username = param("USERNAME")
    password = resolve_password(env)
    if bool(username) != bool(password):
        raise ConfigError("USERNAME and PASSWORD must be set together")
    if ":" in username:
        raise ConfigError("USERNAME must not contain a colon")

    return Config(
        username=username,
        password=password,
        ca_file=validate_ca_file(param("CA_FILE"), site_root(env)),
        base_url=validate_base_url(param("BASE_URL")),
        config_id=config_id,
        tag=tag,
        message_format=message_format,
        verify_tls=verify_tls,
        timeout=timeout,
    )


def clean_text(value: str, limit: int = MAX_ID_CHARS) -> str:
    """Drop control characters (keeping newlines) and bound the length."""
    value = value.replace("\\n", "\n")  # Checkmk escapes newlines in plugin output
    value = "".join(ch for ch in value if ch == "\n" or ch >= " " and ch != "\x7f")
    value = value.strip()
    if len(value) > limit:
        value = value[: limit - 1].rstrip() + "…"
    return value


def parse_event(env: dict[str, str]) -> Event:
    def ctx(name: str, limit: int = MAX_ID_CHARS) -> str:
        return clean_text(env.get("NOTIFY_" + name, ""), limit)

    what = ctx("WHAT").upper()
    if what not in ("HOST", "SERVICE"):
        raise ConfigError("Checkmk event invalid: NOTIFY_WHAT must be HOST or SERVICE")
    host = ctx("HOSTNAME")
    if not host:
        raise ConfigError("Checkmk event invalid: missing NOTIFY_HOSTNAME")
    service = ctx("SERVICEDESC")
    if what == "SERVICE" and not service:
        raise ConfigError("Checkmk event invalid: missing NOTIFY_SERVICEDESC")

    notification_type = ctx("NOTIFICATIONTYPE").upper()
    if not notification_type:
        raise ConfigError("Checkmk event invalid: missing NOTIFY_NOTIFICATIONTYPE")

    prefix = "HOST" if what == "HOST" else "SERVICE"
    return Event(
        what=what,
        notification_type=notification_type,
        host=host,
        address=ctx("HOSTADDRESS"),
        site=ctx("OMD_SITE"),
        state=ctx(prefix + "STATE").upper(),
        output=ctx(prefix + "OUTPUT", MAX_OUTPUT_CHARS),
        service=service if what == "SERVICE" else "",
        long_output=ctx("LONGSERVICEOUTPUT", MAX_LONG_OUTPUT_CHARS) if what == "SERVICE" else "",
        author=ctx("NOTIFICATIONAUTHOR"),
        comment=ctx("NOTIFICATIONCOMMENT", MAX_COMMENT_CHARS),
    )


# --- event -> message ----------------------------------------------------------------------

# Event types that decide the Apprise type regardless of the current state.
EVENT_TYPE_MAP = {
    "RECOVERY": "success",
    "ACKNOWLEDGEMENT": "info",
    "DOWNTIMESTART": "info",
    "DOWNTIMEEND": "info",
    "DOWNTIMECANCELLED": "warning",
    "FLAPPINGSTART": "warning",
    "FLAPPINGSTOP": "info",  # flapping stopped does not imply a healthy state
    "FLAPPINGDISABLED": "info",
    "CUSTOM": "info",
}
STATE_TYPE_MAP = {
    "OK": "success",
    "UP": "success",
    "WARNING": "warning",
    "UNKNOWN": "warning",
    "UNREACHABLE": "warning",
    "CRITICAL": "failure",
    "DOWN": "failure",
}
EVENT_LABELS = {
    "RECOVERY": "RECOVERY",
    "ACKNOWLEDGEMENT": "ACKNOWLEDGEMENT",
    "DOWNTIMESTART": "DOWNTIME START",
    "DOWNTIMEEND": "DOWNTIME END",
    "DOWNTIMECANCELLED": "DOWNTIME CANCELLED",
    "FLAPPINGSTART": "FLAPPING START",
    "FLAPPINGSTOP": "FLAPPING STOP",
    "FLAPPINGDISABLED": "FLAPPING DISABLED",
    "CUSTOM": "CUSTOM",
}


def map_type(event: Event) -> str:
    """Apprise type for an event; the event type wins over the current state."""
    if event.notification_type in EVENT_TYPE_MAP:
        return EVENT_TYPE_MAP[event.notification_type]
    return STATE_TYPE_MAP.get(event.state, "info")


def build_title(event: Event) -> str:
    subject = event.host if event.what == "HOST" else f"{event.host} - {event.service}"
    label = EVENT_LABELS.get(event.notification_type)
    if label is None:  # PROBLEM and anything unknown: lead with the state
        return f"{event.state or event.notification_type}: {subject}"
    state = f" ({event.state})" if event.state else ""
    return f"{label}: {subject}{state}"


def build_body(event: Event, message_format: str = "text") -> str:
    """Render the body as plain text or, for ``html``, as escaped HTML.

    Apprise converts the input format to what each target needs (HTML -> text for
    text-only targets such as Signal, HTML -> Markdown for Markdown targets, HTML
    unchanged for HTML targets). Apprise has no Markdown -> text converter, which is
    why Markdown is not offered. ``html.escape`` neutralizes every monitoring value.
    """
    rich = message_format == "html"

    def lines(pairs: list[tuple[str, str]]) -> str:
        rendered = []
        for label, value in pairs:
            if not value:
                continue
            multiline = value.startswith("\n")
            if rich:
                label = f"<b>{label}:</b>"
                value = html.escape(value, quote=False).replace("\n", "<br>")
            else:
                label = f"{label}:"
            rendered.append(f"{label}{value}" if multiline else f"{label} {value}")
        return ("<br>" if rich else "\n").join(rendered)

    status = [("Service", event.service)] if event.what == "SERVICE" else []
    status += [("State", event.state), ("Output", event.output)]
    if event.long_output and event.long_output != event.output:
        status.append(("Details", "\n" + event.long_output))
    if event.comment:
        # Downtime comments already start with "Author (name): ..."; don't repeat the author
        add_author = event.author and event.author not in event.comment
        author = f" ({event.author})" if add_author else ""
        status.append(("Comment", event.comment + author))
    context = [
        ("Host", event.host),
        ("Address", event.address),
        ("Site", event.site),
        ("Notification", event.notification_type),
    ]
    blocks = [block for block in (lines(status), lines(context)) if block]
    return ("<br><br>" if rich else "\n\n").join(blocks)


def build_payload(config: Config, event: Event) -> dict[str, str]:
    payload = {
        "title": build_title(event),
        "body": build_body(event, config.message_format),
        "type": map_type(event),
        "format": config.message_format,
    }
    if config.tag:
        payload["tag"] = config.tag
    return payload


# --- transport -----------------------------------------------------------------------------


def build_url(config: Config) -> str:
    return f"{config.base_url}/notify/{urllib.parse.quote(config.config_id, safe='')}"


def build_ssl_context(verify: bool, ca_file: str = "") -> ssl.SSLContext:
    """Default context; with ``ca_file`` only that file is trusted (private CA or pinned
    self-signed certificate), the host name is still checked against the certificate."""
    context = ssl.create_default_context(cafile=ca_file or None)
    if not verify:
        context.check_hostname = False
        context.verify_mode = ssl.CERT_NONE
    return context


def classify_status(status: int) -> int:
    """Map an HTTP status to a Checkmk exit code (matrix: docs/TECHNICAL_SPEC.md).

    Only 200 counts as delivered: Apprise answers 200 once the notification is sent.
    Older Apprise API versions answered 204 for a key without (or with an empty)
    configuration, which means nothing was sent, so every other 2xx is a failure too.
    """
    if status == 200:
        return EXIT_OK
    if status in (408, 424, 429) or status >= 500:
        return EXIT_TEMPORARY
    return EXIT_PERMANENT


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: ANN001, ARG002
        return None  # surfaces the 3xx as HTTPError instead of following it


def _status_result(status: int) -> Result:
    code = classify_status(status)
    if code == EXIT_OK:
        return Result(EXIT_OK, f"Apprise notification delivered (HTTP {status})")
    if code == EXIT_TEMPORARY:
        return Result(code, f"Apprise notification temporarily failed: HTTP {status}")
    hint = ""
    if status in (401, 403):
        hint = " - check credentials and the Apprise access mode"
    elif status in (400, 404):
        hint = " - check Config ID, routing tag and Apprise access mode"
    elif 300 <= status < 400:
        hint = " - redirects are not followed, use the final Apprise URL"
    elif 200 <= status < 300:
        hint = " - nothing was sent; check that the Config ID has a saved configuration"
    return Result(code, f"Apprise notification permanently failed: HTTP {status}{hint}")


def _error_result(error: BaseException) -> Result:
    reason = error.reason if isinstance(error, urllib.error.URLError) else error
    if isinstance(reason, ssl.SSLCertVerificationError):
        # Kept temporary on purpose: losing an alert because of a certificate that an
        # administrator may fix within Checkmk's retry window is worse than a retry.
        category = "TLS certificate verification failed (check CA, host name and expiry)"
    elif isinstance(reason, ssl.SSLError):
        category = "TLS error"
    elif isinstance(reason, socket.gaierror):
        category = "DNS resolution failed"
    elif isinstance(reason, TimeoutError):
        category = "timeout"
    else:
        category = "connection error"
    return Result(EXIT_TEMPORARY, f"Apprise notification temporarily failed: {category}")


def send(config: Config, payload: dict[str, str]) -> Result:
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json",
        "User-Agent": USER_AGENT,
    }
    if config.username:
        credentials = f"{config.username}:{config.password}".encode()
        headers["Authorization"] = "Basic " + base64.b64encode(credentials).decode("ascii")
    request = urllib.request.Request(  # noqa: S310 - scheme validated in validate_base_url
        build_url(config),
        data=json.dumps(payload).encode("utf-8"),
        headers=headers,
        method="POST",
    )
    # Proxies from the site environment are ignored on purpose: the request goes
    # directly to the configured Apprise server (deterministic, no implicit proxy).
    handlers: list = [urllib.request.ProxyHandler({}), _NoRedirect()]
    if config.base_url.startswith("https://"):
        context = build_ssl_context(config.verify_tls, config.ca_file)
        handlers.append(urllib.request.HTTPSHandler(context=context))
    opener = urllib.request.build_opener(*handlers)
    # The socket timeout only bounds each single operation (a server that sends one byte
    # per second would never trip it) and does not cover DNS lookups. The request therefore
    # runs in a daemon thread and the configured timeout is a total deadline for it.
    outcome: dict[str, object] = {}

    def worker() -> None:
        try:
            with opener.open(request, timeout=config.timeout) as response:
                response.read(DRAIN_BYTES)
                outcome["status"] = response.status
        except urllib.error.HTTPError as error:
            error.close()
            outcome["status"] = error.code
        except BaseException as error:  # noqa: BLE001 - classified below, text is never printed
            outcome["error"] = error

    thread = threading.Thread(target=worker, daemon=True)
    thread.start()
    thread.join(config.timeout)
    if thread.is_alive():
        return Result(EXIT_TEMPORARY, "Apprise notification temporarily failed: timeout")
    status = outcome.get("status")
    if isinstance(status, int):
        return _status_result(status)
    error = outcome.get("error")
    if isinstance(error, OSError | ValueError):  # URLError, timeouts, TLS, connection resets
        return _error_result(error)
    return Result(EXIT_TEMPORARY, "Apprise notification temporarily failed: protocol error")


def main(env: dict[str, str]) -> int:
    try:
        config = parse_config(env)
        event = parse_event(env)
    except ConfigError as error:
        print(f"Apprise configuration invalid: {error}")
        return EXIT_PERMANENT
    if not config.verify_tls and config.base_url.startswith("https://"):
        print("WARNING: TLS certificate verification is disabled")
    if config.username and config.base_url.startswith("http://"):
        print("WARNING: credentials are sent over unencrypted HTTP")
    result = send(config, build_payload(config, event))
    print(result.message)
    return result.exit_code


if __name__ == "__main__":
    sys.exit(main(dict(os.environ)))
