checkbot/app/checkmk.py

import json
from dataclasses import dataclass
from typing import Sequence

import httpx

from .constants import (CHECKMK_CA_FILE, CHECKMK_SECRET, CHECKMK_SITE, CHECKMK_URL,
                        CHECKMK_USER, CHECKMK_VERIFY_TLS, FLAG_OFF, FLAG_ON)
from .models import HOST_SERVICE_NAME

API_PATH = "/check_mk/api/1.0"
SERVICES_PATH = "/domain-types/service/collections/all"
HOSTS_PATH = "/domain-types/host/collections/all"
VERSION_PATH = "/version"
DOWNTIMES_PATH = "/domain-types/downtime/collections/all"
DOWNTIME_TYPE_PARAM = "downtime_type"
DOWNTIME_TYPE_HOST = "host"
DOWNTIME_TYPE_SERVICE = "service"
DOWNTIME_TYPES = (DOWNTIME_TYPE_HOST, DOWNTIME_TYPE_SERVICE)
DEFAULT_TIMEOUT = 15.0
STATE_OK = 0

COLUMNS = [
    "host_name",
    "description",
    "state",
    "state_type",
    "plugin_output",
    "last_state_change",
    "scheduled_downtime_depth",
    "is_flapping",
    "acknowledged",
]

HOST_COLUMNS = [
    "name",
    "state",
    "plugin_output",
    "last_state_change",
    "scheduled_downtime_depth",
    "acknowledged",
]
# Checkmk: 0 UP, 1 DOWN, 2 UNREACH. Intern DOWN=2 und UNREACH=3, damit min_state-Regeln wie bei Services greifen.
HOST_STATE_MAP = {0: 0, 1: 2, 2: 3}
HOST_STATE_FALLBACK = 2

FAILURE_UNREACHABLE = "unreachable"
FAILURE_UNEXPECTED = "unexpected"

NOT_OK_QUERY = {"op": "!=", "left": "state", "right": str(STATE_OK)}

HOST_QUERY_POST = "post"
HOST_QUERY_GET = "get"

DOWNTIME_MODE_TYPE = "downtime_type"
DOWNTIME_MODE_QUERY = "query"

# Lesbare Hinweise je HTTP-Status, nur für Menschen (nie Teil von .brief, das bleibt wortgleich wie bisher).
STATUS_HINTS = {
    401: "Zugangsdaten prüfen",
    403: "Rechte des Automation-Users prüfen",
    404: "Pfad oder Site prüfen",
    422: "Anfrage von Checkmk abgelehnt (Format/Spalten prüfen)",
}
SERVER_ERROR_HINT = "Checkmk-Serverfehler"


def status_hint(status: int) -> str:
    if status >= 500:
        return SERVER_ERROR_HINT
    return STATUS_HINTS.get(status, "")


class CheckmkError(Exception):
    kind = FAILURE_UNEXPECTED

    def __init__(self, message: str, brief: str = "", hint: str = ""):
        super().__init__(message)
        # Kurztext ohne Antwortinhalt, damit er gefahrlos in Chat-Meldungen steht.
        self.brief = brief or type(self).__name__
        self.hint = hint


class CheckmkUnreachable(CheckmkError):
    kind = FAILURE_UNREACHABLE


@dataclass(frozen=True)
class ServiceState:
    host: str
    service: str
    state: int
    hard: bool
    output: str
    last_state_change: int
    in_downtime: bool
    flapping: bool
    acknowledged: bool = False


@dataclass(frozen=True)
class HostState:
    host: str
    state: int
    output: str
    last_state_change: int
    in_downtime: bool
    acknowledged: bool

    def as_service(self) -> ServiceState:
        return ServiceState(self.host, HOST_SERVICE_NAME, self.state, True, self.output,
                            self.last_state_change, self.in_downtime, False, self.acknowledged)


@dataclass(frozen=True)
class Downtime:
    host: str
    service: str | None
    start: str
    end: str
    comment: str
    author: str


def _parse_downtime(extensions: dict) -> Downtime:
    return Downtime(
        host=extensions.get("host_name", ""),
        service=extensions.get("service_description") or None,
        start=str(extensions.get("start_time", "")),
        end=str(extensions.get("end_time", "")),
        comment=extensions.get("comment", ""),
        author=extensions.get("author", ""),
    )


def _equals(column: str, value: str) -> dict:
    return {"op": "=", "left": column, "right": value}


HARD_STATE_TYPES = frozenset({"1", "hard"})


def _is_hard(value) -> bool:
    return str(value).lower() in HARD_STATE_TYPES


def _parse_service(extensions: dict) -> ServiceState:
    return ServiceState(
        host=extensions["host_name"],
        service=extensions["description"],
        state=int(extensions["state"]),
        hard=_is_hard(extensions.get("state_type", 1)),
        output=extensions.get("plugin_output", ""),
        last_state_change=int(extensions.get("last_state_change", 0)),
        in_downtime=int(extensions.get("scheduled_downtime_depth", 0)) > 0,
        flapping=bool(int(extensions.get("is_flapping", 0))),
        acknowledged=bool(int(extensions.get("acknowledged", 0))),
    )


def _parse_host(extensions: dict) -> HostState:
    return HostState(
        host=extensions["name"],
        state=HOST_STATE_MAP.get(int(extensions["state"]), HOST_STATE_FALLBACK),
        output=extensions.get("plugin_output", ""),
        last_state_change=int(extensions.get("last_state_change", 0)),
        in_downtime=int(extensions.get("scheduled_downtime_depth", 0)) > 0,
        acknowledged=bool(int(extensions.get("acknowledged", 0))),
    )


def extract_extensions(body: dict) -> list[dict]:
    # Checkmk nennt die Collection je Endpunkt mal "value", mal "members".
    rows = body.get("value", body.get("members", []))
    return [item["extensions"] for item in rows]


def _items(body, parser) -> list:
    try:
        return [parser(extensions) for extensions in extract_extensions(body)]
    except (KeyError, TypeError, ValueError, AttributeError) as exc:
        raise CheckmkError(f"Unerwartetes Antwortformat ({type(exc).__name__})",
                           brief="Antwortformat unerwartet") from exc


def downtimes_from_states(services: Sequence[ServiceState], hosts: Sequence[HostState]) -> list[Downtime]:
    host_downtimes = [Downtime(h.host, None, "", "", "", "") for h in hosts if h.in_downtime]
    service_downtimes = [Downtime(s.host, s.service, "", "", "", "") for s in services if s.in_downtime]
    return host_downtimes + service_downtimes


def is_configured(store) -> bool:
    return all(store.get(key) for key in (CHECKMK_URL, CHECKMK_SITE, CHECKMK_USER, CHECKMK_SECRET))


class CheckmkClient:
    def __init__(self, base_url: str, site: str, user: str, secret: str,
                 verify: bool | str = True, timeout: float = DEFAULT_TIMEOUT,
                 transport: httpx.AsyncBaseTransport | None = None):
        self._client = httpx.AsyncClient(
            base_url=f"{base_url.rstrip('/')}/{site}{API_PATH}",
            headers={"Authorization": f"Bearer {user} {secret}", "Accept": "application/json"},
            verify=verify,
            timeout=timeout,
            transport=transport,
        )
        # Merkt sich den zuletzt funktionierenden Weg der Host-Abfrage (siehe problem_hosts_with_fallback).
        self._host_query_mode = HOST_QUERY_POST

    @classmethod
    def from_settings(cls, store) -> "CheckmkClient | None":
        if not is_configured(store):
            return None
        url, site = store.get(CHECKMK_URL), store.get(CHECKMK_SITE)
        user, secret = store.get(CHECKMK_USER), store.get(CHECKMK_SECRET)
        if store.get(CHECKMK_VERIFY_TLS, FLAG_ON) == FLAG_OFF:
            verify: bool | str = False
        else:
            verify = store.get(CHECKMK_CA_FILE) or True
        return cls(url, site, user, secret, verify=verify)

    async def _request(self, method: str, path: str, **kwargs) -> dict:
        try:
            response = await self._client.request(method, path, **kwargs)
        except httpx.HTTPError as exc:
            raise CheckmkUnreachable(f"{type(exc).__name__}: {exc}", brief=type(exc).__name__) from exc
        status = response.status_code
        if status >= 400:
            error = CheckmkUnreachable if status >= 500 else CheckmkError
            raise error(f"HTTP {status} on {path}: {response.text[:200]}", brief=f"HTTP {status}",
                       hint=status_hint(status))
        try:
            return response.json()
        except ValueError as exc:
            raise CheckmkError(f"Antwort auf {path} ist kein JSON", brief="Antwort ist kein JSON") from exc

    async def version(self) -> dict:
        return await self._request("GET", VERSION_PATH)

    async def problem_services(self) -> list[ServiceState]:
        body = await self._request(
            "POST",
            SERVICES_PATH,
            json={"query": NOT_OK_QUERY, "columns": COLUMNS},
        )
        return _items(body, _parse_service)

    async def _fetch_hosts_post(self) -> dict:
        return await self._request("POST", HOSTS_PATH, json={"query": NOT_OK_QUERY, "columns": HOST_COLUMNS})

    async def _fetch_hosts_get(self) -> dict:
        return await self._request("GET", HOSTS_PATH, params={"columns": ",".join(HOST_COLUMNS)})

    async def problem_hosts(self) -> list[HostState]:
        if self._host_query_mode == HOST_QUERY_GET:
            hosts = _items(await self._fetch_hosts_get(), _parse_host)
            return [h for h in hosts if h.state != STATE_OK]
        return _items(await self._fetch_hosts_post(), _parse_host)

    async def _scan_hosts(self) -> tuple[list[dict], list[HostState], str]:
        order = (self._host_query_mode,
                HOST_QUERY_GET if self._host_query_mode == HOST_QUERY_POST else HOST_QUERY_POST)
        last_exc: CheckmkError | None = None
        for mode in order:
            try:
                body = await (self._fetch_hosts_get() if mode == HOST_QUERY_GET else self._fetch_hosts_post())
            except CheckmkError as exc:
                last_exc = exc
                continue
            rows = extract_extensions(body)
            try:
                hosts = _items(body, _parse_host)
            except CheckmkError:
                hosts = []
            if mode == HOST_QUERY_GET:
                rows = [r for r in rows if str(r.get("state", "")) != "0"]
                hosts = [h for h in hosts if h.state != STATE_OK]
            self._host_query_mode = mode
            return rows, hosts, mode
        raise last_exc

    async def problem_hosts_with_fallback(self) -> tuple[list[HostState], str]:
        _, hosts, mode = await self._scan_hosts()
        return hosts, mode

    async def selftest_hosts(self) -> tuple[list[dict], list[HostState], str]:
        return await self._scan_hosts()

    async def raw_services(self) -> dict:
        return await self._request("POST", SERVICES_PATH, json={"query": NOT_OK_QUERY, "columns": COLUMNS})

    async def selftest_services(self) -> tuple[list[dict], list[ServiceState]]:
        body = await self.raw_services()
        rows = extract_extensions(body)
        try:
            services = _items(body, _parse_service)
        except CheckmkError:
            services = []
        return rows, services

    async def sample_service(self) -> dict | None:
        # Die Service-Collection kennt (Stand 2.3) kein Limit/Pagination; um die Antwort klein zu
        # halten, wird nur ohne Zustandsfilter abgefragt und davon ausschließlich das erste Element
        # ausgewertet, statt die komplette Liste zu laden.
        body = await self._request("POST", SERVICES_PATH, json={"columns": COLUMNS})
        rows = extract_extensions(body)
        return rows[0] if rows else None

    async def sample_host(self) -> dict | None:
        try:
            body = await self._request("POST", HOSTS_PATH, json={"columns": HOST_COLUMNS})
        except CheckmkError:
            body = await self._request("GET", HOSTS_PATH, params={"columns": ",".join(HOST_COLUMNS)})
        rows = extract_extensions(body)
        return rows[0] if rows else None

    async def problems(self) -> list[ServiceState]:
        return await self.problem_services()

    async def _services(self, query: dict) -> list[ServiceState]:
        body = await self._request("POST", SERVICES_PATH, json={"query": query, "columns": COLUMNS})
        return _items(body, _parse_service)

    async def service_status(self, host: str, service: str) -> ServiceState | None:
        query = {"op": "and", "expr": [_equals("host_name", host), _equals("description", service)]}
        services = await self._services(query)
        return services[0] if services else None

    async def host_services(self, host: str) -> list[ServiceState]:
        return await self._services(_equals("host_name", host))

    async def downtimes(self, host: str | None = None) -> list[Downtime]:
        downtimes: list[Downtime] = []
        for downtime_type in DOWNTIME_TYPES:
            body = await self._request("GET", DOWNTIMES_PATH, params={DOWNTIME_TYPE_PARAM: downtime_type})
            downtimes += _items(body, _parse_downtime)
        if host:
            wanted = host.casefold()
            downtimes = [d for d in downtimes if d.host.casefold() == wanted]
        return downtimes

    async def _downtimes_by_query(self, host: str | None = None) -> list[Downtime]:
        downtimes: list[Downtime] = []
        for is_service in ("0", "1"):
            query: dict = {"op": "=", "left": "is_service", "right": is_service}
            if host:
                query = {"op": "and", "expr": [query, _equals("host_name", host)]}
            body = await self._request("GET", DOWNTIMES_PATH, params={"query": json.dumps(query)})
            downtimes += _items(body, _parse_downtime)
        if host:
            wanted = host.casefold()
            downtimes = [d for d in downtimes if d.host.casefold() == wanted]
        return downtimes

    async def downtimes_with_fallback(self, host: str | None = None) -> tuple[list[Downtime], str]:
        try:
            return await self.downtimes(host), DOWNTIME_MODE_TYPE
        except CheckmkError:
            return await self._downtimes_by_query(host), DOWNTIME_MODE_QUERY

    async def aclose(self) -> None:
        await self._client.aclose()