#!/usr/bin/env python3
# Copyright (C) 2026 Benjamin Knapp
# SPDX-License-Identifier: GPL-2.0-only
"""Generic JSON API special agent.

Fetches one or more JSON documents over HTTP(S), extracts the configured fields
by path, and prints a single 'json_api' section (one line) merging the results
of every endpoint. Each endpoint carries its own method/headers/auth/timeout and
its own list of extractions; a single '--endpoint' JSON blob is passed per
endpoint, with its secret (if any) supplied out-of-band as '--secret_<i>'.

An endpoint that cannot be fetched does not fail the whole data source: each of
its configured services is emitted as 'not found' with the fetch error, so only
those services go bad while the other endpoints keep reporting.

The section payload looks like:

    {"results": [{"service": "...", "path": "...", "url": "...",
                  "found": true, "value": "UP", "error": null,
                  "levels_upper": ["fixed", [80, 90]], "levels_lower": null,
                  "match": ["must_match", "UP|ok"]}, ...],
     "endpoints": [{"name": "...", "url": "...", "ok": true, "error": null,
                    "status": 200, "elapsed": 0.031, "size": 412,
                    "final_url": "...", "cert_expiry": 1800000000.0,
                    "from_cache": false}, ...]}

The 'endpoints' list carries one record per configured endpoint - the outcome of
the request itself (HTTP status, how long it took, how much came back) - which
becomes the endpoint's own 'JSON API <name>' service, independently of the fields
extracted from the body.
"""

import argparse
import hashlib
import json
import math
import os
import re
import ssl
import sys
import tempfile
import time
from collections.abc import Sequence
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit

import requests
from cmk.utils import password_store as _legacy_pwstore

try:
    # Checkmk 2.5+: public convenience API for secret options.
    from cmk.password_store.v1_unstable import parser_add_secret_option, resolve_secret_option

    _HAVE_PWSTORE_V1 = True
except ImportError:
    # Checkmk 2.4: fall back to the internal password store (same on-disk format).
    _HAVE_PWSTORE_V1 = False

_PATH_TOKEN = re.compile(
    r"\['(?P<sq>[^']*)'\]"  # bracket-quoted key, single quotes: ['foo.bar']
    r"|\[\"(?P<dq>[^\"]*)\"\]"  # bracket-quoted key, double quotes: ["foo.bar"]
    r"|\[(?P<idx>\d+)\]"  # array index: [0]
    r"|(?P<key>[^.\[\]]+)"  # plain dotted key
)


def _add_secret_option(
    parser: argparse.ArgumentParser, name: str, help_text: str, required: bool = True
) -> None:
    """Register a secret option, adapting to the available API.

    Both paths end up reading the same server-side-call rendering: a bare
    ``Secret`` becomes ``--<name>-id "<id>:<password_store_file>"``.
    """
    if _HAVE_PWSTORE_V1:
        parser_add_secret_option(parser, long=f"--{name}", help=help_text, required=required)
    else:
        parser.add_argument(f"--{name}-id", required=required, help=help_text)


def _reveal_secret(args: argparse.Namespace, name: str) -> str:
    """Resolve a secret from the parsed args, across Checkmk 2.4 and 2.5+."""
    if _HAVE_PWSTORE_V1:
        return resolve_secret_option(args, name).reveal()
    secret_id, store_file = getattr(args, f"{name}_id").split(":", 1)
    return _legacy_pwstore.lookup(Path(store_file), secret_id)


def _resolve_path(data: object, path: str) -> tuple[bool, object]:
    """Resolve a dotted path with optional [index] segments.

    Returns (found, value). Supports e.g. 'a.b', 'a[0].b', leading '$.' is
    stripped. Keys that contain '.' or '[' can be addressed with JSONPath-style
    bracket-quoted segments, e.g. "a['foo.bar'].b" or 'a["foo.bar"].b'. Array
    wildcards ('[*]') are handled one level up, in _expand_wildcards.
    """
    current = data
    cleaned = path.strip()
    if cleaned.startswith("$."):
        cleaned = cleaned[2:]
    elif cleaned.startswith("$"):
        cleaned = cleaned[1:]
    for match in _PATH_TOKEN.finditer(cleaned):
        index = match.group("idx")
        if index is not None:
            i = int(index)
            if not isinstance(current, list) or i >= len(current):
                return False, None
            current = current[i]
            continue
        # A dict key, either plain or bracket-quoted. group() returns "" (not
        # None) for an empty quoted key like [''], so test for None explicitly.
        key = match.group("sq")
        if key is None:
            key = match.group("dq")
        if key is None:
            key = match.group("key")
        if not isinstance(current, dict) or key not in current:
            return False, None
        current = current[key]
    return True, current


def parse_arguments(argv: Sequence[str]) -> argparse.Namespace:
    # First pass: learn how many endpoints there are, so we can register one
    # secret option per endpoint index before the real parse.
    pre = argparse.ArgumentParser(add_help=False)
    pre.add_argument("--endpoint", action="append", default=[])
    known, _ = pre.parse_known_args(argv)

    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--endpoint",
        action="append",
        default=[],
        required=True,
        metavar="JSON",
        help="JSON endpoint spec, repeatable. One '--secret_<i>' may accompany each.",
    )
    parser.add_argument(
        "--debug",
        action="store_true",
        help="Write request/response diagnostics to stderr (never into the section on "
        "stdout). Intended for running the program by hand - e.g. the call copied out "
        "of 'cmk -D <host>' - while building or debugging a rule.",
    )
    for index in range(len(known.endpoint)):
        _add_secret_option(
            parser, f"secret_{index}", f"Secret for endpoint {index}", required=False
        )
    return parser.parse_args(argv)


# Cap the response body we buffer and parse, so a monitored endpoint returning
# a huge (or endless) body cannot exhaust memory on the Checkmk server.
_MAX_RESPONSE_BYTES = 50 * 1024 * 1024  # 50 MiB
# Endpoints are fetched concurrently, but with a modest ceiling.
_MAX_FETCH_WORKERS = 8


class _ResponseTooLarge(Exception):
    """The response body exceeded ``_MAX_RESPONSE_BYTES``."""


def _read_capped(response: requests.Response, limit: int) -> bytes:
    """Read the response body, refusing to buffer more than ``limit`` bytes."""
    chunks: list[bytes] = []
    total = 0
    for chunk in response.iter_content(chunk_size=65536):
        total += len(chunk)
        if total > limit:
            raise _ResponseTooLarge(f"Response exceeds the {limit}-byte limit")
        chunks.append(chunk)
    return b"".join(chunks)


def _effective_body(endpoint: dict) -> object | None:
    """The request body actually sent with this endpoint's request.

    A request body only makes sense for POST; never smuggle one onto a GET. Both
    the Content-Type defaulting in ``_build_session`` and the request itself in
    ``_fetch`` route through here, so the two can never disagree about whether a
    body is present (e.g. defaulting Content-Type for a body that is then never
    sent).
    """
    return endpoint.get("body") if endpoint.get("method", "GET") == "POST" else None


def _apply_proxy(session: requests.Session, endpoint: dict) -> None:
    """Configure the session's HTTP proxy from the endpoint's 'proxy' spec.

    The server-side call resolves the rule's Proxy choice into one of:
      {"mode": "url", "url": "http://proxy:3128"} - route via this proxy
      {"mode": "no_proxy"}                        - ignore any environment proxy
      {"mode": "environment"} / absent            - honour HTTP(S)_PROXY from
                                                    the monitoring host's env
    An explicit proxy URL takes precedence over the environment; 'no_proxy'
    turns off requests' env-based proxy lookup for this request.
    """
    proxy = endpoint.get("proxy")
    if not isinstance(proxy, dict):
        return
    match proxy.get("mode"):
        case "url" if proxy.get("url"):
            session.proxies = {"http": proxy["url"], "https": proxy["url"]}
        case "no_proxy":
            session.trust_env = False


def _build_session(endpoint: dict, secret: str | None) -> tuple[requests.Session, dict[str, str]]:
    session = requests.Session()
    _apply_proxy(session, endpoint)
    headers = {name: value for name, value in endpoint.get("headers", [])}
    match endpoint.get("auth"):
        case "auth_login":
            session.auth = (endpoint["username"], secret or "")
        case "auth_token":
            headers["Authorization"] = "Bearer " + (secret or "")
    if _effective_body(endpoint) is not None and not any(
        h.lower() == "content-type" for h in headers
    ):
        headers["Content-Type"] = "application/json"
    return session, headers


# Body preview length in --debug output: enough to see the shape of a response
# without dumping a whole (possibly huge) document to the terminal.
_DEBUG_BODY_PREVIEW = 4000


def _debug(enabled: bool, message: str) -> None:
    """Write one diagnostic line to stderr when --debug is set.

    Kept strictly on stderr so a --debug run stays parseable by Checkmk (the
    section still goes to stdout untouched) while a consultant watches the
    request/response detail on the terminal.
    """
    if enabled:
        sys.stderr.write(f"[json_api debug] {message}\n")


def _redacted_headers(headers: dict[str, str]) -> dict[str, str]:
    """Headers with the Authorization value masked, for safe debug output.

    Bearer tokens live in the Authorization header; basic-auth credentials live
    on ``session.auth`` (never in ``headers``), so they cannot leak here.
    """
    return {
        name: ("<redacted>" if name.lower() == "authorization" else value)
        for name, value in headers.items()
    }


def _accepted_statuses(endpoint: dict) -> set[int]:
    """Extra HTTP status codes to accept beyond 2xx, from the endpoint config.

    Some APIs signal state through the status code - e.g. a '/health' endpoint
    that answers 503 with a JSON body describing what is down. Listing those
    codes here lets the agent read and extract that body instead of failing the
    whole endpoint. 2xx is always accepted; this only widens the set.
    """
    return {code for code in endpoint.get("accept_status") or [] if isinstance(code, int)}


def _verify_arg(endpoint: dict) -> bool | str:
    """The ``verify`` value for requests: the configured flag, or a CA-bundle path.

    ``verify_cert`` is the operator's explicit on/off toggle (off is the
    documented, insecure opt-out; see the ruleset help). A custom CA bundle lets
    a private-CA endpoint be verified without turning verification off, so it
    only applies when verification is on. The disabled value is returned
    straight from the config rather than as a literal, so this stays a single
    source of truth for the toggle.
    """
    verify = endpoint.get("verify_cert", True)
    ca_bundle = endpoint.get("ca_bundle")
    if verify and ca_bundle:
        return ca_bundle
    return verify


def _client_cert(endpoint: dict) -> str | tuple[str, str] | None:
    """The ``cert`` value for requests (mutual TLS): None, certfile, or
    (certfile, keyfile) when the key lives in a separate file."""
    cert = endpoint.get("client_cert")
    if not isinstance(cert, dict) or not cert.get("cert"):
        return None
    key = cert.get("key")
    return (cert["cert"], key) if key else cert["cert"]


def _peer_cert_expiry(response: requests.Response) -> float | None:
    """The peer certificate's ``notAfter`` as a Unix epoch, or ``None``.

    Read off the still-open socket: the request is made with ``stream=True`` (so
    ``_read_capped`` can bound the body), which means the connection is live at
    this point and no second handshake is needed.

    It does reach through ``requests`` into urllib3's pooled connection, which is
    not a public API, and there are ordinary reasons for it to come up empty - a
    reused connection from the pool may not expose the socket, ``verify=False``
    yields an empty cert dict, and plain HTTP has no certificate at all. So every
    failure degrades to ``None`` and the endpoint service simply reports no
    certificate, rather than the agent breaking over a best-effort extra.
    """
    try:
        connection = getattr(response.raw, "connection", None)
        sock = getattr(connection, "sock", None)
        cert = sock.getpeercert() if sock is not None else None
    except Exception:  # noqa: BLE001 - a best-effort extra must never fail a fetch
        return None
    if not isinstance(cert, dict):
        return None
    not_after = cert.get("notAfter")
    if not isinstance(not_after, str):
        return None
    try:
        return float(ssl.cert_time_to_seconds(not_after))
    except ValueError:
        return None
# Per-endpoint response cache. Opt-in: no TTL configured means "always fetch",
# which is the right default for monitoring. It exists for rate-limited,
# expensive or fanned-out endpoints, where the request RATE is the problem rather
# than the freshness. Checkmk's own fetcher cache cannot express this - it is
# host-wide, all-or-nothing, and sized by a site-global setting during checking.
_CACHE_DIR_NAME = "json_api_cache"
# Files this old are removed on the next write, so a rule edit that changes an
# endpoint's identity (and thus its key) cannot leak files forever.
_CACHE_PRUNE_AFTER = 7 * 86400


def _cache_dir() -> Path | None:
    """The directory to cache responses in, or ``None`` if there isn't one.

    Prefers the tmp directory Checkmk hands the agent, then the site's own tmp,
    and only then the system default - a monitoring cache belongs inside the site
    so it is cleaned out with it. ``None`` (nothing writable) disables caching
    rather than failing the fetch.
    """
    directory = Path(tempfile.gettempdir()) / _CACHE_DIR_NAME
    if mk_tmp := os.environ.get("MK_TMPDIR"):
        directory = Path(mk_tmp) / _CACHE_DIR_NAME
    elif omd_root := os.environ.get("OMD_ROOT"):
        directory = Path(omd_root) / "tmp" / _CACHE_DIR_NAME
    try:
        directory.mkdir(parents=True, exist_ok=True)
    except OSError:
        return None
    return directory


def _cache_key(endpoint: dict) -> str:
    """A stable hash over everything that changes this endpoint's response.

    The secret is deliberately NOT part of it and is never stored: it does not
    reach this blob at all (it travels separately as '--secret_<i>'), and hashing
    it would put a credential-derived value on disk for no benefit. Two endpoints
    differing only by credential are a pathological config; the identity below
    covers the realistic cases.
    """
    identity = json.dumps(
        [
            endpoint.get("url"),
            endpoint.get("method", "GET"),
            _effective_body(endpoint),
            endpoint.get("headers"),
            endpoint.get("auth"),
            endpoint.get("verify_cert", True),
            endpoint.get("ca_bundle"),
            endpoint.get("client_cert"),
            endpoint.get("accept_status"),
            endpoint.get("proxy"),
        ],
        sort_keys=True,
        default=str,
    )
    return hashlib.sha256(identity.encode("utf-8")).hexdigest()


def _cache_ttl(endpoint: dict) -> float | None:
    """The configured cache TTL in seconds, or ``None`` for "always fetch"."""
    ttl = endpoint.get("cache_ttl")
    if isinstance(ttl, bool) or not isinstance(ttl, (int, float)):
        return None
    return float(ttl) if ttl > 0 else None


def _cache_read(endpoint: dict, ttl: float) -> tuple[bytes, dict] | None:
    """A cached ``(body, meta)`` younger than ``ttl``, else ``None``.

    Anything unreadable or malformed counts as a miss, so a broken cache file
    costs one extra request and nothing else.
    """
    directory = _cache_dir()
    if directory is None:
        return None
    path = directory / f"{_cache_key(endpoint)}.json"
    try:
        entry = json.loads(path.read_text(encoding="utf-8"))
        stored = float(entry["stored"])
        body = entry["body"].encode("utf-8")
        meta = entry["meta"]
    except (OSError, ValueError, KeyError, TypeError, AttributeError):
        return None
    if not isinstance(meta, dict):
        return None
    age = time.time() - stored
    # A negative age means the clock moved backwards; treat it as a miss rather
    # than serving something we cannot reason about.
    if age < 0 or age > ttl:
        return None
    # 'elapsed' is dropped deliberately: no request was made, so there is no
    # response time. Replaying the original one would chart a measurement that
    # never happened, over and over, for the whole TTL. Status, size and the final
    # URL DO still describe the body being served, so they are kept.
    return body, {**meta, "elapsed": None, "from_cache": True, "cache_age": age}


def _cache_write(endpoint: dict, body: bytes, meta: dict) -> None:
    """Store a fresh response. Best effort: a failure must not fail the fetch."""
    directory = _cache_dir()
    if directory is None:
        return
    path = directory / f"{_cache_key(endpoint)}.json"
    entry = {
        "stored": time.time(),
        "body": body.decode("utf-8", "replace"),
        "meta": {k: v for k, v in meta.items() if k not in ("from_cache", "cache_age")},
    }
    try:
        # Write-then-rename so a concurrent read never sees a half-written file,
        # and 0600 because a response body can hold anything the API returns.
        temporary = path.with_suffix(f".{os.getpid()}.tmp")
        temporary.write_text(json.dumps(entry), encoding="utf-8")
        temporary.chmod(0o600)
        temporary.replace(path)
        _prune_cache(directory)
    except OSError:
        return


def _prune_cache(directory: Path) -> None:
    """Drop cache files nothing will read again (stale keys after a rule edit)."""
    cutoff = time.time() - _CACHE_PRUNE_AFTER
    try:
        entries = list(directory.glob("*.json"))
    except OSError:
        return
    for entry in entries:
        try:
            if entry.stat().st_mtime < cutoff:
                entry.unlink(missing_ok=True)
        except OSError:
            continue


def _fetch(
    endpoint: dict, secret: str | None, debug: bool = False
) -> tuple[object | None, str | None, dict]:
    """Fetch one endpoint: ``(document, error, meta)``.

    ``meta`` describes the request itself - HTTP status, wall-clock duration,
    body size and the final URL after any redirects - and is filled in on every
    exit path (a failed request still took time). It feeds the endpoint's own
    service, so 'the API answered, but slowly' is visible without configuring a
    field for it.
    """
    meta: dict[str, object] = {
        "status": None,
        "elapsed": None,
        "size": None,
        "final_url": None,
        "cert_expiry": None,
        "from_cache": False,
        "cache_age": None,
    }
    # A cached body younger than the endpoint's TTL is served without touching the
    # network at all - that is the whole point, for an API with a request quota.
    if (ttl := _cache_ttl(endpoint)) is not None and (hit := _cache_read(endpoint, ttl)):
        cached_body, cached_meta = hit
        _debug(debug, f"  served from cache ({cached_meta['cache_age']:.0f}s old)")
        try:
            return json.loads(cached_body), None, cached_meta
        except ValueError as exc:
            # Cached something unparseable: fall through and fetch fresh.
            _debug(debug, f"  cached body is not valid JSON ({exc}), fetching")
    started = time.monotonic()

    def _timed() -> dict:
        meta["elapsed"] = time.monotonic() - started
        return meta

    session, headers = _build_session(endpoint, secret)
    timeout = endpoint.get("timeout")
    method = endpoint.get("method", "GET")
    allow_redirects = endpoint.get("follow_redirects", True)
    accepted = _accepted_statuses(endpoint)
    body = _effective_body(endpoint)
    if debug:
        _debug(debug, f"{method} {endpoint.get('url', '?')}")
        for name, value in _redacted_headers(headers).items():
            _debug(debug, f"  header {name}: {value}")
        if session.auth is not None:
            _debug(debug, "  basic auth: <redacted>")
        if body is not None:
            _debug(debug, f"  body: {body}")
    try:
        response = session.request(
            method,
            endpoint["url"],
            data=body,
            headers=headers,
            verify=_verify_arg(endpoint),
            cert=_client_cert(endpoint),
            allow_redirects=allow_redirects,
            timeout=timeout if timeout is not None else 30.0,
            stream=True,  # so _read_capped can bound the buffered body
        )
        with response:
            status = response.status_code
            meta["status"] = status
            meta["final_url"] = response.url
            # While the socket is still open and before the body is read. Done
            # here rather than after the status checks so a certificate is still
            # reported for an endpoint answering 503 - the cert is a property of
            # the connection, not of the response code.
            meta["cert_expiry"] = _peer_cert_expiry(response)
            # With redirects disabled a 3xx is not an error to requests, but it
            # is not the JSON we asked for - report it clearly (this is the SSRF
            # hardening path, where a silent "not valid JSON" would mislead).
            if not allow_redirects and 300 <= status < 400:
                location = response.headers.get("Location", "?")
                _debug(debug, f"  HTTP {response.status_code} redirect to {location} (blocked)")
                return (
                    None,
                    f"Unexpected {status} redirect to {location} (redirects disabled)",
                    _timed(),
                )
            # 2xx is always fine; extra codes can be opted in per endpoint so an
            # API that signals via the status code (e.g. 503 + a JSON health
            # body) can still be read. Anything else fails, carrying the code.
            if not (200 <= status < 300 or status in accepted):
                reason = getattr(response, "reason", "") or ""
                return None, f"HTTP {status}{f' {reason}' if reason else ''}", _timed()
            raw = _read_capped(response, _MAX_RESPONSE_BYTES)
    except _ResponseTooLarge as exc:
        _debug(debug, f"  {exc}")
        return None, str(exc), _timed()
    except requests.exceptions.RequestException as exc:
        _debug(debug, f"  request failed: {exc}")
        return None, f"Request failed: {exc}", _timed()

    meta["size"] = len(raw)
    if debug:
        preview = raw[:_DEBUG_BODY_PREVIEW].decode("utf-8", "replace")
        suffix = " ...(truncated)" if len(raw) > _DEBUG_BODY_PREVIEW else ""
        _debug(debug, f"  HTTP {response.status_code}, {len(raw)} bytes")
        _debug(debug, f"  body: {preview}{suffix}")

    try:
        document = json.loads(raw)
    except ValueError as exc:
        return None, f"Response is not valid JSON: {exc}", _timed()
    timed = _timed()
    if ttl is not None:
        # Only a response we could actually parse is worth caching; an error or a
        # non-JSON body would just be replayed for the whole TTL.
        _cache_write(endpoint, raw, timed)
    return document, None, timed


_WILDCARD = "[*]"

# The 'element' of a leaf that has no element behind it, because the wildcard's
# container was missing altogether. A distinct sentinel rather than ``None``:
# ``None`` is also an ordinary JSON value, so an array holding one (``[null]``)
# would otherwise be indistinguishable from "there was no array at all".
_NO_ELEMENT = object()


def _split_wildcards(path: str) -> list[str]:
    """Split a path on every '[*]' wildcard into segments.

    A path with N wildcards yields N+1 segments: the part before the first
    wildcard, the part between each consecutive pair, and the trailing value
    path. A leading '.' is stripped from every segment after the first. A path
    with no wildcard yields a single-element list (the whole path).

    'pods[*].containers[*].ready' -> ['pods', 'containers', 'ready']
    'nodes[*].health'             -> ['nodes', 'health']
    'items[*]'                    -> ['items', '']
    'status'                      -> ['status']
    """
    parts = path.split(_WILDCARD)
    return [parts[0]] + [p[1:] if p.startswith(".") else p for p in parts[1:]]


def _matches_filter(element: object, filt: object) -> bool:
    """Whether a wildcard/count element satisfies the configured filter.

    The filter path is resolved WITHIN the element and the resolved scalar is
    compared per the operator (equals / not_equals / regex / not_regex, the
    regexes being full matches). A missing path, a non-scalar value, or a bad
    regex all count as "no match" (the element is dropped), so a filter only
    ever keeps elements it can positively confirm - including for the negated
    operators, where an element lacking the field is dropped rather than kept.
    """
    if not isinstance(filt, dict) or not filt.get("path"):
        return True  # no filter configured → keep everything
    found, value = _resolve_path(element, filt["path"])
    if not found:
        return False
    text = _label_value(value)
    if text is None:
        return False
    target = filt.get("value", "")
    match filt.get("op"):
        case "not_equals":
            return text != target
        case "regex":
            try:
                return re.fullmatch(target, text) is not None
            except re.error:
                return False
        case "not_regex":
            try:
                return re.fullmatch(target, text) is None
            except re.error:
                return False
        case _:  # "equals" (the default)
            return text == target


def _aggregate_mode(spec: dict) -> str | None:
    """The configured aggregation ('count'/'sum'/'avg'/'min'/'max'), or None.

    ``count: true`` is the superseded boolean form of ``aggregate: "count"``: a
    rule saved before the aggregate dropdown still carries it (the ruleset
    migrates it only when the rule is next opened in Setup), and so may a
    hand-written '--endpoint' blob. Reading it here keeps both working.
    """
    mode = spec.get("aggregate")
    if isinstance(mode, str) and mode:
        return mode
    return "count" if spec.get("count") else None


def _as_float(value: object) -> float | None:
    """A JSON value as a finite float, or None when it is not numeric.

    Numeric strings are accepted (APIs do quote numbers); booleans are not (they
    are states, and summing them is never what was meant).
    """
    if isinstance(value, bool):
        return None
    if isinstance(value, (int, float)):
        number = float(value)
    elif isinstance(value, str):
        try:
            number = float(value)
        except ValueError:
            return None
    else:
        return None
    return number if math.isfinite(number) else None


_AGGREGATE_NOT_NUMERIC = (
    "element is not numeric (use a '[*]' path to aggregate a field inside each element)"
)


def _aggregate_numbers(mode: str, values: list[object]) -> tuple[bool, object, str]:
    """Reduce ``values`` with ``mode`` into ``(found, value, error)``.

    Every value must be numeric; one that is not fails the whole aggregation
    rather than being skipped, so a wrong path is visible instead of quietly
    changing the result. An empty input sums to 0 (a filter that matched nothing
    legitimately totals zero) but has no average, minimum or maximum.
    """
    numbers: list[float] = []
    for value in values:
        number = _as_float(value)
        if number is None:
            return False, None, _AGGREGATE_NOT_NUMERIC
        numbers.append(number)
    if not numbers:
        if mode == "sum":
            return True, 0, ""
        return False, None, "no elements to aggregate"
    match mode:
        case "sum":
            return True, _trim(sum(numbers)), ""
        case "avg":
            return True, _trim(sum(numbers) / len(numbers)), ""
        case "min":
            return True, _trim(min(numbers)), ""
        case "max":
            return True, _trim(max(numbers)), ""
    return False, None, f"unknown aggregation '{mode}'"


def _trim(number: float) -> float | int:
    """An integral result as an int, so a sum of integers reads as '15', not '15.0'.

    Every value goes through ``_as_float`` to be aggregated, which makes even a
    sum of plain integers a float; the check renders the value as it arrives, so
    trimming here is what keeps the service summary looking like the JSON did.
    """
    return int(number) if number.is_integer() else number


def _aggregate_container(mode: str, container: object, filt: object) -> tuple[bool, object, str]:
    """Aggregate the elements of the array/object at a wildcard-free path.

    'count' is the number of elements (array length or number of object keys);
    the numeric functions reduce the elements themselves, so this form fits a
    collection of numbers - a collection of objects wants a '[*]' path naming the
    field to aggregate. A scalar has nothing to aggregate and is surfaced as a
    misconfiguration rather than, say, counting the characters of a string.
    """
    pairs = _iter_container(container)
    if pairs is None:
        return False, None, "value at path is not a list or object (cannot aggregate)"
    elements = [element for _default, element in pairs]
    if isinstance(filt, dict) and filt.get("path"):
        # Aggregate only the matching elements (e.g. how many nodes are NOT 'ok').
        elements = [element for element in elements if _matches_filter(element, filt)]
    if mode == "count":
        return True, len(elements), ""
    return _aggregate_numbers(mode, elements)


def _aggregate_leaves(
    mode: str,
    leaves: list[tuple[list[str], bool, object, str, object]],
    filt: object,
) -> tuple[bool, object, str]:
    """Aggregate the leaves of a '[*]' expansion into one value.

    This is the counterpart of fanning a wildcard out into one service per
    element: 'nodes[*].load' with 'avg' yields a single service holding the
    average load instead of one service per node. The filter applies to the
    element the leaf came from, exactly as it does when fanning out.

    A leaf whose value path is missing is skipped, so every mode - 'count'
    included - sees the same set of elements: the ones that actually have the
    field. 'nodes[*].load' counts the nodes reporting a load, which is what
    naming the field asks for, and keeps 'count' consistent with the average
    over those same values. If NO leaf resolved at all that is reported instead -
    a mistyped value path must not read as an empty collection.
    """
    # A missing container yields exactly one leaf, with no element behind it -
    # identified by the sentinel, so an array holding a single JSON null (a real
    # element, with a real length of 1) is not mistaken for it.
    if len(leaves) == 1 and leaves[0][4] is _NO_ELEMENT:
        return False, None, leaves[0][3]
    filtering = isinstance(filt, dict) and bool(filt.get("path"))
    kept = [
        (found, value)
        for _labels, found, value, _error, element in leaves
        if not filtering or _matches_filter(element, filt)
    ]
    values = [value for found, value in kept if found]
    if kept and not values:
        return False, None, "path not found in any element"
    if mode == "count":
        # Counting needs no numbers, only elements - a collection of strings has
        # a length just as much as one of numbers does.
        return True, len(values), ""
    return _aggregate_numbers(mode, values)


# Checkmk host names are used in file paths, Livestatus queries and config, so
# only this conservative set survives; anything else becomes '_'. Mirrors what
# Checkmk itself accepts for a host name.
_HOST_NAME_INVALID = re.compile(r"[^-0-9A-Za-z_.]")


def _piggyback_host(element: object, host_path: object) -> str | None:
    """The piggyback host name for one element, or ``None`` when there is none.

    The name is read from a field WITHIN the element (the same way a filter or a
    label is), then sanitised: a host name ends up in file paths and config, so
    the character set is restricted rather than trusted. An element whose field is
    missing, non-scalar, or sanitises to nothing yields ``None`` - the caller
    leaves that element's services on the polling host rather than inventing a
    host name for them.
    """
    if not isinstance(host_path, str) or not host_path.strip():
        return None
    found, value = _resolve_path(element, host_path.strip())
    if not found:
        return None
    text = _label_value(value)
    if text is None:
        return None
    cleaned = _HOST_NAME_INVALID.sub("_", text.strip()).strip("_.")
    return cleaned or None


def _result(
    spec: dict,
    service: str,
    found: bool,
    value: object,
    error: str,
    url: str,
    labels: list[dict] | None = None,
    host: str | None = None,
) -> dict:
    # Values that are not JSON scalars (dict/list) are reported as strings.
    if found and isinstance(value, (dict, list)):
        value = json.dumps(value)
    return {
        "service": service,
        # Which host this result belongs to: None is the polling host (its
        # services go into the plain section), a name means a piggyback host and
        # is stripped back out before the section is written.
        "host": host,
        "path": spec["path"],
        "url": url,
        "found": found,
        "value": value if found else None,
        "error": None if found else error,
        "levels_upper": spec.get("levels_upper"),
        "levels_lower": spec.get("levels_lower"),
        "match": spec.get("match"),
        "calc": spec.get("calc"),
        "unit": spec.get("unit"),
        # Passed through for the check: it derives the per-second rate / the age
        # (it owns the value store and knows "now").
        "value_as": spec.get("value_as"),
        # Already applied here; carried along only so the check can say so in the
        # service's Details.
        "aggregate": _aggregate_mode(spec),
        "labels": labels or [],
    }


def _iter_container(container: object) -> list[tuple[str, object]] | None:
    """(default_label, element) pairs for a wildcard-expandable container.

    A JSON array expands element-by-element with the index as the default label;
    a JSON object (map) expands entry-by-entry with the key as the default label
    - so a Spring Boot Actuator '/health' 'components' map, keyed by component
    name, iterates just like an array. Returns None for a scalar (or a missing
    node), letting the caller report 'array or object not found'.
    """
    if isinstance(container, list):
        return [(str(index), element) for index, element in enumerate(container)]
    if isinstance(container, dict):
        return [(str(key), value) for key, value in container.items()]
    return None


def _expand_wildcards(
    node: object, segments: list[str], label_path: str | None
) -> list[tuple[list[str], bool, object, str, object]]:
    """Recursively expand the wildcard segments of a path under ``node``.

    ``segments`` is the output of :func:`_split_wildcards`: one entry per
    wildcard level plus the trailing value path. Returns one
    ``(label_segments, found, value, error, element)`` tuple per leaf - i.e. per
    element of the cartesian product of all wildcard levels. ``label_segments``
    carries one label per wildcard level (joined into the composite service name
    by the caller); ``label_path`` is resolved relative to each level's element,
    with the array index (or object key) as the fallback. ``element`` is the leaf
    element node (``_NO_ELEMENT`` when the container was missing) so the caller
    can resolve per-element service labels against it.
    """
    # Base case: no more wildcards, ``segments[0]`` is the value path itself.
    if len(segments) == 1:
        value_path = segments[0]
        if value_path:
            found, value = _resolve_path(node, value_path)
        else:
            found, value = True, node
        return [([], found, value, "path not found in element", node)]

    container_path, *rest = segments
    found_container, container = _resolve_path(node, container_path)
    pairs = _iter_container(container) if found_container else None
    if pairs is None:
        return [([], False, None, "array or object not found at wildcard path", _NO_ELEMENT)]

    expanded: list[tuple[list[str], bool, object, str, object]] = []
    elements = [element for _default, element in pairs]
    for label, element in zip(_element_labels(pairs, label_path), elements):
        for sub_labels, found, value, error, leaf in _expand_wildcards(element, rest, label_path):
            expanded.append(([label, *sub_labels], found, value, error, leaf))
    return expanded


def _extract(document: object, extractions: list[dict], url: str) -> list[dict]:
    results = []
    for spec in extractions:
        label_specs = spec.get("labels") or []
        segments = _split_wildcards(spec["path"])
        aggregate = _aggregate_mode(spec)
        filt = spec.get("filter")
        if len(segments) == 1:
            found, value = _resolve_path(document, spec["path"])
            error = "path not found in response"
            # Aggregating a wildcard-free path reduces the array/object it holds.
            if found and aggregate:
                found, value, error = _aggregate_container(aggregate, value, filt)
            labels = _resolve_labels(label_specs, document)
            results.append(_result(spec, spec["service"], found, value, error, url, labels))
            continue

        leaves = _expand_wildcards(document, segments, spec.get("label_path"))

        # Aggregating a '[*]' path collapses the expansion into ONE service (the
        # sum/average/... over the elements) instead of fanning it out. Service
        # labels then have no single element to resolve against, so they come
        # from the response root.
        if aggregate:
            found, value, error = _aggregate_leaves(aggregate, leaves, filt)
            labels = _resolve_labels(label_specs, document)
            results.append(_result(spec, spec["service"], found, value, error, url, labels))
            continue

        # One or more array wildcards: one service per cartesian-product
        # element, labelled by a ' / '-joined composite (one segment per level).
        # An optional filter keeps only the elements whose sub-field matches
        # (e.g. only the nodes that are NOT 'ok'); a not-found leaf is kept so
        # the "array or object not found" error still surfaces.
        host_path = spec.get("piggyback_host")
        for label_segments, found, value, error, element in leaves:
            if found and isinstance(filt, dict) and filt.get("path"):
                if not _matches_filter(element, filt):
                    continue
            # With a piggyback host field, the element becomes its own Checkmk
            # host, so the host carries the identity and the service keeps its
            # plain name - 'JSON Health' on 50 hosts, not 'JSON Health node-01'
            # on one. An element whose host name does not resolve falls back to
            # the labelled name on the polling host, so it is still monitored
            # instead of vanishing.
            host = _piggyback_host(element, host_path)
            label = " / ".join(label_segments)
            if host is not None:
                service = spec["service"]
            else:
                service = f"{spec['service']} {label}" if label else spec["service"]
            labels = _resolve_labels(label_specs, element)
            results.append(_result(spec, service, found, value, error, url, labels, host))
    return results


def _label_value(value: object) -> str | None:
    """A JSON scalar as a label string, or ``None`` for null/object/array.

    Booleans render as JSON ('true'/'false') to match the value rendering; a
    non-scalar makes no sense as a label value and is skipped by the caller.
    """
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, (int, float)):
        return str(value)
    if isinstance(value, str):
        return value
    return None


def _label_key_from_path(path: str) -> str:
    """The last dict-key segment of a path, used as the default label key.

    'metadata.name' -> 'name', "data['foo.bar']" -> 'foo.bar', 'components[*]' ->
    'components'. Falls back to the cleaned path when it ends in an array index.
    """
    cleaned = path.strip()
    if cleaned.startswith("$."):
        cleaned = cleaned[2:]
    elif cleaned.startswith("$"):
        cleaned = cleaned[1:]
    cleaned = cleaned.replace("[*]", "")  # a wildcard is not part of the key
    key = None
    for match in _PATH_TOKEN.finditer(cleaned):
        token = match.group("sq")
        if token is None:
            token = match.group("dq")
        if token is None:
            token = match.group("key")
        if token is not None:
            key = token
    return key or cleaned


def _resolve_labels(label_specs: list[dict], element: object) -> list[dict]:
    """Resolve SERVICE-label specs relative to ``element`` into ``{key, value}``.

    ``element`` is each '[*]' element (or the whole document for a non-wildcard
    extraction), so a per-element service gets its own label value. Unresolved
    paths and non-scalar / null values are skipped, so a label is emitted only
    when it has a value. The 'json_api/' key namespace is added by the check.
    """
    resolved = []
    for spec in label_specs:
        path = spec.get("path")
        if not path:
            continue
        found, value = _resolve_path(element, path)
        if not found:
            continue
        text = _label_value(value)
        if text is None:
            continue
        resolved.append({"key": spec.get("key") or _label_key_from_path(path), "value": text})
    return resolved


def _resolve_host_labels(label_specs: list[dict], document: object) -> dict:
    """Resolve HOST-label specs from the ``document`` root into ``{key: value}``.

    Host labels are host-wide, so they attach to no service and are resolved once
    per endpoint response. Two shapes are supported:

    * plain path (no '[*]'): one label ``<key>: <scalar at path>`` where ``key``
      is the given key or the path's last segment.
    * wildcard path (contains '[*]', e.g. ``components[*]``): one label PER
      element, keyed ``<key>/<element-id>`` so keys stay unique (a host label map
      cannot repeat a key); the value comes from ``value_field`` resolved within
      each element, defaulting to ``'true'`` (set-membership tags).

    Later keys win on collision. The 'json_api/' namespace is added by the check.
    """
    labels: dict[str, str] = {}
    for spec in label_specs:
        path = spec.get("path")
        if not path:
            continue
        base_key = spec.get("key") or _label_key_from_path(path)
        value_field = spec.get("value_field") or ""
        segments = _split_wildcards(path)
        if len(segments) == 1:
            found, value = _resolve_path(document, path)
            text = _label_value(value) if found else None
            if text is not None:
                labels[base_key] = text
            continue
        # Wildcard: one unique label per element.
        for label_segments, _found, _value, _error, element in _expand_wildcards(
            document, segments, None
        ):
            if element is None:
                continue
            suffix = "/".join(str(part) for part in label_segments)
            key = f"{base_key}/{suffix}" if suffix else base_key
            if value_field:
                found, value = _resolve_path(element, value_field)
                text = _label_value(value) if found else None
                if text is None:
                    continue  # value field missing / non-scalar → skip this element
            else:
                text = "true"
            labels[key] = text
    return labels


def _element_labels(pairs: list[tuple[str, object]], label_path: str | None) -> list[str]:
    """One label per (default_label, element) pair, guaranteed unique.

    The label comes from label_path within each element, falling back to the
    default label (an array index or object key). If a label value repeats
    across elements, every occurrence of it is suffixed with its position, so
    two elements can never collapse into one service.
    """
    labels = []
    for position, (default_label, element) in enumerate(pairs):
        if label_path:
            found, value = _resolve_path(element, label_path)
            labels.append(str(value) if found else default_label)
        else:
            labels.append(default_label)
    counts: dict[str, int] = {}
    for label in labels:
        counts[label] = counts.get(label, 0) + 1
    return [
        f"{label} [{index}]" if counts[label] > 1 else label for index, label in enumerate(labels)
    ]


def _url_without_query(url: str) -> str:
    """``url`` with its query string and fragment removed.

    An API that authenticates via a query parameter (``?api_key=...``) would
    otherwise carry that secret into a service description, which travels much
    further than the service details do: notifications, availability reports, BI
    aggregations, the metric paths on disk. The full URL stays in the endpoint
    service's details, where it belongs.

    Anything that does not parse - or that parses to nothing at all - is returned
    unchanged rather than replaced by an empty item.
    """
    try:
        split = urlsplit(url)
    except ValueError:
        return url
    return urlunsplit(split._replace(query="", fragment="")) or url


def _endpoint_name(endpoint: dict, url: str) -> str:
    """The endpoint's service item: its configured name, else the bare URL."""
    name = endpoint.get("name")
    if isinstance(name, str) and name.strip():
        return name.strip()
    return _url_without_query(url)


def _endpoint_record(
    endpoint: dict, url: str, ok: bool, error: str | None, meta: dict | None = None
) -> dict:
    """One 'endpoints' entry: the outcome of the request itself."""
    meta = meta or {}
    return {
        "name": _endpoint_name(endpoint, url),
        "url": url,
        "ok": ok,
        "error": error,
        "status": meta.get("status"),
        "elapsed": meta.get("elapsed"),
        "size": meta.get("size"),
        "final_url": meta.get("final_url"),
        "cert_expiry": meta.get("cert_expiry"),
        "from_cache": bool(meta.get("from_cache")),
        "cache_age": meta.get("cache_age"),
    }


def _process_endpoint(
    args: argparse.Namespace, index: int, endpoint: dict
) -> tuple[list[dict], dict, dict]:
    """Fetch one endpoint: its (extraction results, host labels, own record).

    Any failure is confined to this endpoint: secret resolution, the fetch, and
    the extraction (including a malformed endpoint blob, e.g. one missing 'url')
    all fall back to a 'not found' result carrying the error, so the endpoint's
    own services go bad while the rest of the section stays intact. A failed
    endpoint contributes no host labels, and its record carries the error for its
    own service.
    """
    url = endpoint.get("url", "?")
    debug = getattr(args, "debug", False)
    _debug(debug, f"endpoint {index}: {url}")

    def _fail(error: str, meta: dict | None = None) -> tuple[list[dict], dict, dict]:
        results = [
            _result(
                {"path": spec.get("path", "?")}, spec.get("service", "?"), False, None, error, url
            )
            for spec in endpoint.get("extractions", [])
            if isinstance(spec, dict)
        ]
        # Guarantee the failure is visible even when the blob carries no usable
        # extractions to hang it on (otherwise it would silently vanish).
        return (
            results or [_result({"path": "?"}, url, False, None, error, url)],
            {},
            _endpoint_record(endpoint, url, False, error, meta),
        )

    try:
        secret = _reveal_secret(args, f"secret_{index}") if endpoint.get("auth") else None
    except Exception as exc:  # e.g. a stale/missing password-store reference
        return _fail(f"Secret resolution failed: {exc}")

    try:
        document, error, meta = _fetch(endpoint, secret, debug)
        if error is not None:
            return _fail(error, meta)
        results = _extract(document, endpoint.get("extractions", []), url)
        host_labels = _resolve_host_labels(endpoint.get("host_labels", []), document)
        if debug:
            for result in results:
                outcome = (
                    f"found: {result['value']!r}"
                    if result["found"]
                    else f"NOT FOUND: {result['error']}"
                )
                _debug(debug, f"  service {result['service']!r} <- {result['path']} -> {outcome}")
            for key, value in host_labels.items():
                _debug(debug, f"  host label json_api/{key} = {value}")
        return (results, host_labels, _endpoint_record(endpoint, url, True, None, meta))
    except Exception as exc:  # malformed endpoint blob, unexpected extraction error, ...
        return _fail(f"Endpoint processing failed: {exc}")


def _split_by_host(results: list[dict]) -> tuple[list[dict], dict[str, list[dict]]]:
    """Partition results into the polling host's own and per-piggyback-host.

    The ``host`` key is an internal routing hint, not part of the section format,
    so it is stripped here: the check sees identical payloads either way and needs
    no knowledge of piggybacking at all. Insertion order is preserved so the
    output stays deterministic across runs.
    """
    own: list[dict] = []
    piggybacked: dict[str, list[dict]] = {}
    for result in results:
        host = result.pop("host", None)
        if host is None:
            own.append(result)
        else:
            piggybacked.setdefault(host, []).append(result)
    return own, piggybacked


def main(argv: Sequence[str] | None = None) -> int:
    args = parse_arguments(sys.argv[1:] if argv is None else argv)

    endpoints = [json.loads(raw) for raw in args.endpoint]
    results: list[dict] = []
    host_labels: dict[str, str] = {}
    endpoint_records: list[dict] = []
    if endpoints:
        # Fetch endpoints concurrently so total runtime is the slowest endpoint,
        # not the sum. pool.map preserves input order, keeping the merged section
        # (and thus service-name disambiguation) deterministic across runs.
        with ThreadPoolExecutor(max_workers=min(len(endpoints), _MAX_FETCH_WORKERS)) as pool:
            per_endpoint = pool.map(
                lambda item: _process_endpoint(args, item[0], item[1]),
                enumerate(endpoints),
            )
            for endpoint_results, endpoint_host_labels, record in per_endpoint:
                results.extend(endpoint_results)
                host_labels.update(endpoint_host_labels)  # later endpoints win per key
                endpoint_records.append(record)

    own, piggybacked = _split_by_host(results)
    # The polling host's own section: the endpoint records live here too, because
    # they describe the REQUEST, which belongs to the host holding the rule - not
    # to any element the response happened to contain.
    payload = {"results": own, "host_labels": host_labels, "endpoints": endpoint_records}
    sys.stdout.write("<<<json_api:sep(0)>>>\n")
    sys.stdout.write(json.dumps(payload) + "\n")
    for host, host_results in piggybacked.items():
        # One section per piggyback host, in the same format - so the check parses
        # it with no idea it was piggybacked, and every field feature (levels,
        # match, counter, timestamp, ...) works there unchanged.
        sys.stdout.write(f"<<<<{host}>>>>\n")
        sys.stdout.write("<<<json_api:sep(0)>>>\n")
        sys.stdout.write(json.dumps({"results": host_results, "host_labels": {}}) + "\n")
    if piggybacked:
        # Close the last piggyback section, or everything after it in the agent
        # output would be attributed to that host.
        sys.stdout.write("<<<<>>>>\n")
    # Flush explicitly so the section is delivered no matter how stdout is
    # connected. Checkmk always reads the agent through a pipe (block-buffered,
    # flushed at exit), but a consultant copying the program call out of
    # `cmk -D <host>` and running it by hand on a TTY otherwise sees nothing
    # until the buffer happens to flush — making a working agent look broken.
    sys.stdout.flush()
    return 0


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