#!/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 Callable, Sequence
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from urllib.parse import quote, quote_plus, urljoin, 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 _strip_root(path: str) -> str:
    """``path`` without surrounding space and without a JSONPath '$' / '$.' root."""
    cleaned = path.strip()
    if cleaned.startswith("$."):
        return cleaned[2:]
    if cleaned.startswith("$"):
        return cleaned[1:]
    return cleaned


def _token_key(match: re.Match[str]) -> str | None:
    """The dict key a path token matched, or ``None`` when it is an array index.

    Tested against ``None`` rather than for truthiness: a bracket-quoted key can
    legitimately be empty (``['']``).
    """
    for group in ("sq", "dq", "key"):
        if (key := match.group(group)) is not None:
            return key
    return None


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
    for match in _PATH_TOKEN.finditer(_strip_root(path)):
        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
        key = _token_key(match)
        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 _read_reportable(response: requests.Response, limit: int) -> tuple[bytes, bool]:
    """Up to ``limit`` bytes of the body, and whether there was more after them.

    Unlike ``_read_capped`` an oversized body is not an error here: the report is
    capped by design, so this stops reading at the limit rather than buffering a
    50 MiB error page in order to show 2 KB of it. Used only where the body is
    wanted *for the report alone* - where it also has to be parsed, it has to be
    read whole anyway.
    """
    chunks: list[bytes] = []
    total = 0
    for chunk in response.iter_content(chunk_size=8192):
        chunks.append(chunk)
        total += len(chunk)
        if total > limit:
            break
    return b"".join(chunks)[:limit], total > limit


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 _auth_header_name(endpoint: dict) -> str | None:
    """The header an 'API key in a header' endpoint puts its key into."""
    if endpoint.get("auth") != "auth_header":
        return None
    name = endpoint.get("auth_header")
    return name if isinstance(name, str) and name.strip() else "X-API-Key"


def _auth_query_name(endpoint: dict) -> str | None:
    """The query parameter an 'API key in a query parameter' endpoint uses."""
    if endpoint.get("auth") != "auth_query":
        return None
    name = endpoint.get("auth_query")
    return name if isinstance(name, str) and name.strip() else "api_key"


def _auth_params(endpoint: dict, secret: str | None) -> dict[str, str] | None:
    """Query parameters carrying the API key, or ``None``.

    Kept out of ``_build_session`` (and out of the endpoint's URL) so the key is
    added by requests at send time only: the configured URL - the one that names
    the service and is printed in debug output - never contains it.
    """
    name = _auth_query_name(endpoint)
    return {name: secret or ""} if name else None


class _Session(requests.Session):
    """A session that also strips an API-key HEADER on a cross-host redirect.

    ``requests`` already does this for ``Authorization`` - that is what protects
    the bearer-token mode - but its ``rebuild_auth`` knows only that one header
    name. An API key lives in a header the *API* names, so without this a
    monitored endpoint that redirects to another host is handed the key in full,
    which is precisely what the password-store modes exist to prevent. Redirects
    are followed by default, so this must be the default too.

    A key in a query parameter needs no equivalent: the redirect's Location
    replaces the query string rather than carrying it along.
    """

    def __init__(self, secret_header: str | None = None) -> None:
        super().__init__()
        self._secret_header = secret_header

    def rebuild_auth(self, prepared_request: object, response: object) -> None:
        super().rebuild_auth(prepared_request, response)  # type: ignore[arg-type]
        if not self._secret_header:
            return
        if self.should_strip_auth(response.request.url, prepared_request.url):  # type: ignore[attr-defined]
            # Headers are a case-insensitive mapping, so the configured spelling
            # need not match what was actually sent.
            prepared_request.headers.pop(self._secret_header, None)  # type: ignore[attr-defined]


def _build_session(
    endpoint: dict, secret: str | None, access_token: str | None = None
) -> tuple[requests.Session, dict[str, str]]:
    session = _Session(_auth_header_name(endpoint))
    _apply_proxy(session, endpoint)
    headers = dict(endpoint.get("headers", []))
    match endpoint.get("auth"):
        case "auth_login":
            session.auth = (endpoint["username"], secret or "")
        case "auth_token":
            headers["Authorization"] = "Bearer " + (secret or "")
        case "auth_oauth2":
            # Already exchanged for a token by the caller; from here it is an
            # ordinary bearer token, so requests' own Authorization stripping on
            # a cross-host redirect protects it like any other.
            headers["Authorization"] = "Bearer " + (access_token or "")
        case "auth_header":
            # The API names the header ('X-API-Key', 'PRIVATE-TOKEN', ...); the
            # value comes from the password store, never from the config.
            headers[_auth_header_name(endpoint) or "X-API-Key"] = 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")


_REDACTED = "<redacted>"


def _redacted_headers(headers: dict[str, str], secret_header: str | None = None) -> dict[str, str]:
    """Headers with credential values 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. An API
    key lives in a header the *API* names, which is why the name to mask has to
    be passed in - masking only 'Authorization' would print the key verbatim.
    """
    masked = {"authorization"}
    if secret_header:
        masked.add(secret_header.lower())
    return {
        name: (_REDACTED if name.lower() in masked else value) for name, value in headers.items()
    }


def _redact_secret(text: str, secret: str | None) -> str:
    """``text`` with every occurrence of ``secret`` masked.

    Used for the 'API key in a query parameter' mode, the one authentication
    style whose credential ends up inside a URL: it is then echoed back by
    ``response.url`` (reported as the endpoint's final URL) and quoted in the
    message of any ``requests`` exception. Both are printed to the terminal and
    stored in the agent output on disk, so the key is stripped out of them here.

    The percent-encoded forms are masked too, because a key containing reserved
    characters reaches the wire encoded.
    """
    if not secret:
        return text
    for form in (secret, quote(secret, safe=""), quote_plus(secret)):
        if form and form in text:
            text = text.replace(form, _REDACTED)
    return text


# Response headers that carry a credential of their own. 'Set-Cookie' is a live
# session token: reporting it in a service's details would put it into the
# monitoring history, and from there into anything that reads a service's output.
_SENSITIVE_RESPONSE_HEADERS = frozenset(
    {"set-cookie", "set-cookie2", "authorization", "proxy-authorization"}
)

# Hard ceiling for the reported body, whatever the rule asks for. The text ends
# up in the endpoint service's details, which are stored with every check result
# and shipped in notifications - this is not the place for a megabyte of JSON.
_MAX_REPORTED_BYTES = 65536
_DEFAULT_REPORTED_BYTES = 2048


def _report_spec(endpoint: dict) -> dict | None:
    """The endpoint's 'report the raw response' settings, or None when off."""
    spec = endpoint.get("show_response")
    return spec if isinstance(spec, dict) else None


def _reported_limit(spec: dict) -> int:
    """How many bytes of the body to report, clamped to the hard ceiling."""
    raw = spec.get("max_bytes")
    limit = int(raw) if isinstance(raw, (int, float)) and not isinstance(raw, bool) else 0
    return max(1, min(limit or _DEFAULT_REPORTED_BYTES, _MAX_REPORTED_BYTES))


def _reported_headers(headers: object, secret: str | None) -> dict[str, str]:
    """The response headers as reported: credential-bearing ones masked.

    The secret is stripped from every value as well - an API that echoes the key
    it was given (in a 'WWW-Authenticate' challenge, say) must not have it stored
    with the check result.
    """
    if not isinstance(headers, dict):
        return {}
    return {
        str(name): (
            _REDACTED
            if str(name).lower() in _SENSITIVE_RESPONSE_HEADERS
            else _redact_secret(str(value), secret)
        )
        for name, value in headers.items()
    }


def _store_report(
    endpoint: dict, meta: dict, raw: bytes, secret: str | None, complete: bool = True
) -> None:
    """Put the reported body (and headers) for ``raw`` into ``meta``, if asked.

    ``complete`` says whether ``raw`` is the whole body. It is not when the body
    was read for the report alone and stopped at the limit, and then the real
    length is unknown - the service says the body was cut off without claiming
    a size it never measured.

    Both are capped and stripped of the secret HERE, in the only function that
    has the whole body, so the untruncated bytes never travel any further. A body
    cut at the limit can split a multi-byte character, which decodes to U+FFFD
    rather than failing the whole report.

    Not stored in the response cache (see _cache_write): the report is derived
    from the body, and the body IS cached, so it is rebuilt on a cache hit
    against the settings in force now rather than the ones a previous check ran
    with.
    """
    spec = _report_spec(endpoint)
    if spec is None:
        return
    limit = _reported_limit(spec)
    meta["body"] = _redact_secret(raw[:limit].decode("utf-8", "replace"), secret)
    meta["body_truncated"] = len(raw) > limit or not complete
    meta["body_size"] = len(raw) if complete else None
    if spec.get("headers", True):
        meta["headers_reported"] = _reported_headers(meta.get("headers"), secret)


# How much of the JSON context to put into ONE field service's details. Smaller
# than the raw-response default on purpose: this text is repeated on every field
# service of the endpoint, not written once to the endpoint's own service.
_DEFAULT_CONTEXT_BYTES = 1024


def _context_spec(endpoint: dict) -> dict | None:
    """The endpoint's 'report the JSON context in the field services' settings."""
    spec = endpoint.get("field_context")
    return spec if isinstance(spec, dict) else None


def _resolve_parent(data: object, path: str) -> tuple[bool, object]:
    """Resolve everything but the LAST segment of a path: the value's container.

    A single-segment path has the document itself as its container, which is the
    right answer: the context of a top-level field IS the response.
    """
    tokens = list(_PATH_TOKEN.finditer(_strip_root(path)))
    if len(tokens) <= 1:
        return True, data
    current = data
    for match in tokens[:-1]:
        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
        key = _token_key(match)
        if not isinstance(current, dict) or key not in current:
            return False, None
        current = current[key]
    return True, current


def _context_object(spec: dict, document: object, path: str, element: object) -> object:
    """The JSON a field's details should show, per the configured source.

    'response' is the whole document. 'element' is the JSON the value was
    actually read from: the current '[*]' element where there is one, else the
    object containing the value (its parent). An aggregation has no single
    element and a '@header.' path is not in the body at all, so both fall back to
    the document - the alternative would be to report nothing where the context
    is arguably most useful.
    """
    if spec.get("source") == "response":
        return document
    if element is not None and element is not _NO_ELEMENT:
        return element
    if _WILDCARD in path or path.strip().startswith(_HEADER_PREFIX):
        return document
    found, parent = _resolve_parent(document, path)
    return parent if found else document


def _context_text(
    spec: dict | None,
    document: object,
    path: str,
    element: object = None,
    secret: str | None = None,
) -> str | None:
    """The pretty-printed, capped, secret-stripped JSON context, or None.

    Pretty-printed rather than verbatim: the document has been parsed by the time
    a field is extracted, and an indented object is what makes the details worth
    reading. The cap is applied to the ENCODED text, so it bounds what is stored
    with every check result; a cut that splits a multi-byte character decodes to
    U+FFFD rather than failing the whole report.
    """
    if spec is None:
        return None
    raw = spec.get("max_bytes")
    limit = int(raw) if isinstance(raw, (int, float)) and not isinstance(raw, bool) else 0
    limit = max(1, min(limit or _DEFAULT_CONTEXT_BYTES, _MAX_REPORTED_BYTES))
    try:
        text = json.dumps(_context_object(spec, document, path, element), indent=2, default=str)
    except (TypeError, ValueError):
        return None
    encoded = _redact_secret(text, secret).encode("utf-8", "replace")
    if len(encoded) <= limit:
        return encoded.decode("utf-8", "replace")
    shown = encoded[:limit].decode("utf-8", "replace")
    return f"{shown}\n... (truncated at {limit} of {len(encoded)} bytes)"


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(name: str = _CACHE_DIR_NAME) -> Path | None:
    """A cache directory of the given name, 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.

    ``name`` separates the response cache from the OAuth2 token cache: they are
    keyed differently and expire differently, so they must not share a directory
    (the pruner walks a whole directory).
    """
    directory = Path(tempfile.gettempdir()) / name
    if mk_tmp := os.environ.get("MK_TMPDIR"):
        directory = Path(mk_tmp) / name
    elif omd_root := os.environ.get("OMD_ROOT"):
        directory = Path(omd_root) / "tmp" / name
    try:
        directory.mkdir(parents=True, exist_ok=True)
    except OSError:
        return None
    return directory


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

    The credential is part of the identity, because for the API-key modes it is
    the ordinary case rather than a pathological one: several rules can poll the
    SAME multi-tenant URL with the same header name and a different key each, and
    the response they get back is per-key. Without it they would share one cache
    file and serve each other's tenant data for the whole TTL.

    Only a hash of the secret is used, never the secret: the cache key ends up in
    a filename, and a SHA-256 of a credential is not one. The secret itself still
    reaches neither the endpoint blob nor the disk.
    """
    identity = json.dumps(
        [
            endpoint.get("url"),
            endpoint.get("method", "GET"),
            _effective_body(endpoint),
            endpoint.get("headers"),
            endpoint.get("auth"),
            # The header / query parameter an API key goes into - a name, not a
            # credential.
            endpoint.get("auth_header"),
            endpoint.get("auth_query"),
            hashlib.sha256(secret.encode("utf-8")).hexdigest() if secret else None,
            endpoint.get("verify_cert", True),
            endpoint.get("ca_bundle"),
            endpoint.get("client_cert"),
            endpoint.get("accept_status"),
            endpoint.get("proxy"),
            # The cached body is the MERGED one, so how the pages were followed
            # is part of what it is: a changed collection path or page limit must
            # not be answered from a cache built under the old settings.
            endpoint.get("pagination"),
        ],
        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, secret: str | None = None) -> 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, secret)}.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' and 'attempts' are dropped deliberately: no request was made, so
    # there is no response time and nothing was retried. Replaying either would
    # report a measurement that never happened, over and over, for the whole TTL -
    # and a stale 'attempts' would hold the endpoint service at the state
    # configured for "a retry was needed" across checks that made no request at
    # all. Status, size and the final URL DO still describe the body being
    # served, so they are kept.
    return body, {**meta, "elapsed": None, "attempts": 1, "from_cache": True, "cache_age": age}


def _cache_write(endpoint: dict, body: bytes, meta: dict, secret: str | None = None) -> 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, secret)}.json"
    entry = {
        "stored": time.time(),
        "body": body.decode("utf-8", "replace"),
        # Facts about the REQUEST, not about the body: replaying them on a cache
        # hit would describe a request that never happened. The reported body and
        # headers are dropped too - they are rebuilt from the cached body on a
        # hit, so a rule that changed its reporting settings meanwhile is honoured.
        "meta": {
            k: v
            for k, v in meta.items()
            if k
            not in (
                "from_cache",
                "cache_age",
                "attempts",
                "body",
                "body_truncated",
                "body_size",
                "headers_reported",
            )
        },
    }
    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


# An access token is refreshed this many seconds BEFORE it expires, so a token
# that is valid when we check cannot expire in flight on a slow request.
_TOKEN_EXPIRY_SKEW = 60.0
# A provider that reports no lifetime at all: assume a short one rather than
# caching a token forever.
_TOKEN_DEFAULT_TTL = 300.0
_TOKEN_CACHE_DIR_NAME = "json_api_token_cache"


class _TokenError(Exception):
    """The token could not be obtained; the endpoint fails with this message."""


def _oauth2_spec(endpoint: dict) -> dict | None:
    """The endpoint's OAuth2 config, or ``None`` when it uses another auth mode."""
    if endpoint.get("auth") != "auth_oauth2":
        return None
    spec = endpoint.get("oauth2")
    return spec if isinstance(spec, dict) and spec.get("token_url") else None


def _token_cache_key(spec: dict, secret: str | None) -> str:
    """A hash over everything that decides WHICH token this is.

    Deliberately not the response cache's key: a token is shared by every request
    that authenticates the same way, and its lifetime comes from the provider's
    'expires_in' rather than from the endpoint's cache TTL. Two endpoints of the
    same rule pointing at the same provider with the same client should reuse one
    token, not fetch two.

    The secret is hashed, never stored - the key becomes a filename.
    """
    identity = json.dumps(
        [
            spec.get("token_url"),
            spec.get("client_id"),
            spec.get("scope"),
            spec.get("audience"),
            spec.get("client_auth", "basic"),
            hashlib.sha256(secret.encode("utf-8")).hexdigest() if secret else None,
        ],
        sort_keys=True,
        default=str,
    )
    return hashlib.sha256(identity.encode("utf-8")).hexdigest()


def _token_cache_path(spec: dict, secret: str | None) -> Path | None:
    directory = _cache_dir(_TOKEN_CACHE_DIR_NAME)
    if directory is None:
        return None
    return directory / f"{_token_cache_key(spec, secret)}.json"


def _cached_token(spec: dict, secret: str | None) -> str | None:
    """A cached access token that is still valid, else ``None``."""
    path = _token_cache_path(spec, secret)
    if path is None:
        return None
    try:
        entry = json.loads(path.read_text(encoding="utf-8"))
        token = entry["token"]
        expires_at = float(entry["expires_at"])
    except (OSError, ValueError, KeyError, TypeError):
        return None
    if not isinstance(token, str) or not token:
        return None
    if time.time() < expires_at:
        return token
    # Expired: drop it now rather than leaving a dead bearer credential on disk
    # until the 7-day sweep. Tokens live for minutes, so that sweep - written for
    # response bodies whose key changed - is far too slow to be the only cleanup.
    try:
        path.unlink(missing_ok=True)
    except OSError:
        pass
    return None


def _store_token(spec: dict, secret: str | None, token: str, ttl: float) -> None:
    """Cache a fresh token. Best effort: failing to cache must not fail the fetch."""
    path = _token_cache_path(spec, secret)
    if path is None:
        return
    entry = {"token": token, "expires_at": time.time() + max(ttl - _TOKEN_EXPIRY_SKEW, 0.0)}
    try:
        temporary = path.with_suffix(f".{os.getpid()}.tmp")
        # An access token is a bearer credential at rest. Storing one is the
        # whole point of caching it - the alternative is asking the identity
        # provider on every check - and it cannot be hashed (it has to be sent)
        # or encrypted usefully (the key would have to sit beside it). Checkmk's
        # own password store and the Graph client in cmk/plugins/emailchecks
        # keep credentials on disk the same way. Mitigated by 0600 and by living
        # in the site's tmp, which is cleared with the site.
        # (CodeQL flags this as clear-text storage. The suppression comment
        # below is honoured by the CodeQL CLI but not by GitHub's default code
        # scanning, so the alert also has to be dismissed in the Security tab.)
        temporary.write_text(  # codeql[py/clear-text-storage-sensitive-data]
            json.dumps(entry), encoding="utf-8"
        )
        temporary.chmod(0o600)
        temporary.replace(path)
        _prune_cache(path.parent)
    except OSError:
        return


def _forget_token(spec: dict, secret: str | None) -> None:
    """Drop the cached token, so the next attempt fetches a fresh one."""
    path = _token_cache_path(spec, secret)
    if path is None:
        return
    try:
        path.unlink(missing_ok=True)
    except OSError:
        return


def _request_token(endpoint: dict, spec: dict, secret: str | None, debug: bool) -> tuple[str, float]:
    """Exchange the client credentials for an access token: ``(token, ttl)``.

    Raises ``_TokenError`` with a message fit for the service, never leaking the
    secret: a provider that rejects the credentials tends to echo the request
    back, so the body is not reported.
    """
    data: dict[str, str] = {"grant_type": "client_credentials"}
    if scope := spec.get("scope"):
        data["scope"] = str(scope)
    if audience := spec.get("audience"):
        data["audience"] = str(audience)

    session = requests.Session()
    _apply_proxy(session, endpoint)
    if spec.get("client_auth") == "post":
        data["client_id"] = str(spec.get("client_id", ""))
        data["client_secret"] = secret or ""
    else:
        session.auth = (str(spec.get("client_id", "")), secret or "")

    timeout = endpoint.get("timeout")
    _debug(debug, f"  fetching an access token from {spec['token_url']}")
    try:
        response = session.post(
            spec["token_url"],
            data=data,
            # The token endpoint is part of the trust chain: verification follows
            # the endpoint's own TLS settings rather than being relaxed here.
            verify=_verify_arg(endpoint),
            cert=_client_cert(endpoint),
            timeout=timeout if timeout is not None else 30.0,
        )
    except requests.exceptions.RequestException as exc:
        # Only the exception TYPE and the token URL, never the message. requests
        # can quote the request it was making, and with the credentials sent in
        # the body that request literally contains the client secret - redacting
        # it afterwards is a weaker guarantee than never assembling it. The
        # token URL is configuration, not a credential.
        raise _TokenError(
            f"Token request to {spec['token_url']} failed ({type(exc).__name__})"
        ) from exc
    with response:
        if not 200 <= response.status_code < 300:
            reason = getattr(response, "reason", "") or ""
            raise _TokenError(
                f"Token request returned HTTP {response.status_code}"
                f"{f' {reason}' if reason else ''}"
                " (check the client credentials, the scope, and how the client "
                "credentials are sent)"
            )
        try:
            payload = response.json()
        except ValueError as exc:
            raise _TokenError(f"Token response is not valid JSON: {exc}") from exc
    if not isinstance(payload, dict):
        raise _TokenError("Token response is not a JSON object")
    token = payload.get("access_token")
    if not isinstance(token, str) or not token:
        raise _TokenError("Token response carries no 'access_token'")
    ttl = _as_float(payload.get("expires_in"))
    return token, ttl if ttl is not None and ttl > 0 else _TOKEN_DEFAULT_TTL


def _access_token(endpoint: dict, spec: dict, secret: str | None, debug: bool) -> str:
    """A valid access token, from the cache when there is one."""
    if (cached := _cached_token(spec, secret)) is not None:
        _debug(debug, "  using the cached access token")
        return cached
    token, ttl = _request_token(endpoint, spec, secret, debug)
    _store_token(spec, secret, token, ttl)
    return token


# Whatever the retry count, the agent never waits longer than this in total. A
# check that sleeps for minutes is a worse failure than the one it is papering
# over: Checkmk kills a special agent that overruns.
_MAX_RETRY_SLEEP = 30.0


def _retry_policy(endpoint: dict) -> tuple[int, float]:
    """``(retries, backoff)`` for this endpoint; ``(0, 0.0)`` when off."""
    retry = endpoint.get("retry")
    if not isinstance(retry, dict):
        return 0, 0.0
    attempts = retry.get("attempts")
    if isinstance(attempts, bool) or not isinstance(attempts, int) or attempts < 1:
        return 0, 0.0
    backoff = retry.get("backoff")
    if isinstance(backoff, bool) or not isinstance(backoff, (int, float)) or backoff < 0:
        backoff = 0.0
    return attempts, float(backoff)


def _retryable_status(status: int) -> bool:
    """Whether repeating a request that answered ``status`` could help.

    A 5xx is the server saying it failed, and a 429 is it saying "later" - both
    can differ on the next attempt. Everything else (a 4xx, a blocked redirect)
    is a decision about the request itself and would answer exactly the same,
    so retrying it only burns the budget.
    """
    return status == 429 or 500 <= status < 600


# A paginated collection arrives one page at a time, and the agent asks for one
# page - so a '[*]' expansion or an aggregation over it silently describes the
# FIRST page and nothing else. 'count' over a queue that pages at 25 reports 25
# however long the queue is: an answer that is wrong without saying so, which is
# worse than an error. Following the pages is opt-in per endpoint because it
# costs requests, but it is the only way that answer becomes true.
#
# However the rule is written, the agent never makes more than this many
# requests for one endpoint. A check that walks a thousand pages is an outage of
# its own - Checkmk kills a special agent that overruns - and a hand-written
# rule (or a blob built by something other than the ruleset) is not bound by the
# ruleset's own range.
_MAX_PAGES = 100
_DEFAULT_MAX_PAGES = 10


def _pagination_spec(endpoint: dict) -> dict | None:
    """The endpoint's 'follow pagination' settings, or None when off."""
    spec = endpoint.get("pagination")
    return spec if isinstance(spec, dict) else None


def _next_source(spec: dict) -> tuple[str, str] | None:
    """Where the next page's URL comes from: ``("body", <path>)`` or ``("link_header", "")``.

    The CascadingSingleChoice value as the rule stores it (a list once it has
    been through JSON). Anything else - a missing choice, an empty body path -
    means pagination cannot be followed at all, which is reported as None rather
    than guessed at.
    """
    raw = spec.get("next")
    if not isinstance(raw, (list, tuple)) or len(raw) != 2:
        return None
    mode, value = raw
    if mode == "body" and isinstance(value, str) and value.strip():
        return "body", _strip_root(value)
    if mode == "link_header":
        return "link_header", ""
    return None


def _items_path(spec: dict) -> str | None:
    """The path to the collection each page carries, '' for the response root.

    '$' (or '$.') is how a response that IS the array says so; ``_strip_root``
    turns it into the empty path, which ``_resolve_path`` resolves to the whole
    document.
    """
    raw = spec.get("items")
    return _strip_root(raw) if isinstance(raw, str) else None


def _pagination_limits(spec: dict) -> tuple[int, int | None]:
    """``(max pages, max elements)``, the page count clamped to ``_MAX_PAGES``."""
    raw = spec.get("max_pages")
    pages = (
        int(raw)
        if isinstance(raw, (int, float)) and not isinstance(raw, bool) and raw >= 1
        else _DEFAULT_MAX_PAGES
    )
    raw = spec.get("max_elements")
    elements = (
        int(raw)
        if isinstance(raw, (int, float)) and not isinstance(raw, bool) and raw >= 1
        else None
    )
    return min(pages, _MAX_PAGES), elements


def _container_len(container: object) -> int:
    """How many elements the merged collection holds so far."""
    return len(container) if isinstance(container, (list, dict)) else 0


def _next_candidate(source: tuple[str, str], document: object, headers: object) -> str | None:
    """The next page's URL as this page states it, or None for "there is none".

    'There is none' is how pagination ENDS, so an absent field, a JSON ``null``
    and an empty string all mean the same thing: the last page has been read.
    """
    mode, path = source
    if mode == "body":
        found, value = _resolve_path(document, path)
        if found and isinstance(value, str) and value.strip():
            return value.strip()
        return None
    # RFC 8288: Link: <https://api/jobs?page=2>; rel="next", <...>; rel="last".
    # Parsed with requests' own parser rather than a regex of ours - it is the
    # same code that fills response.links, and the header is fiddlier than it
    # looks (several links per header, quoted parameters, relative URLs).
    found, header = _resolve_header(headers, f"{_HEADER_PREFIX}Link")
    if not found or not isinstance(header, str):
        return None
    for link in requests.utils.parse_header_links(header):
        if link.get("rel") == "next" and (url := link.get("url", "").strip()):
            return url
    return None


def _next_target(
    candidate: str, current_url: str, origin: str, seen: set[str]
) -> tuple[str | None, str | None]:
    """``(url to fetch, reason not to)`` for a next-page link.

    Relative links are the norm ('/api/v1/jobs?page=2'), so the candidate is
    resolved against the page it came from.

    Two links are refused rather than followed. One on ANOTHER HOST would make
    the response body decide where the Checkmk server sends an authenticated
    request - the same SSRF shape the 'follow redirects' switch exists to
    close - and the endpoint's credentials travel with every page. One already
    fetched means the API is pointing at itself, which would otherwise spend the
    whole page budget re-reading one page. Both stop pagination with a reason the
    endpoint's own service reports; neither fails the endpoint, so the data that
    WAS read still monitors the API.
    """
    url = urljoin(current_url, candidate)
    parsed = urlsplit(url)
    if parsed.scheme.lower() not in ("http", "https") or not parsed.netloc:
        return None, f"the next page link is not an http(s) URL ({candidate})"
    if parsed.netloc.lower() != origin:
        return None, f"the next page link points to another host ({parsed.netloc})"
    if url in seen:
        return None, "the API repeated a page link (pagination loop)"
    return url, None


def _merge_page(container: object, addition: object) -> str | None:
    """Append one page's collection to the first page's, in place; error if any.

    In place because ``container`` is the object inside the first page's document
    that every extraction will read: extending it is what makes the whole
    collection visible to a '[*]' wildcard and an aggregation, while the rest of
    the document (a 'total', a 'generated_at') stays the first page's.

    A JSON object pages by key rather than by position, so the two container
    kinds the rest of the agent already treats alike are both merged - and a page
    whose collection is neither, or is not the same kind as the first page's, is
    an error: silently skipping it would under-count exactly the way unfollowed
    pagination does.
    """
    if isinstance(container, list) and isinstance(addition, list):
        container.extend(addition)
        return None
    if isinstance(container, dict) and isinstance(addition, dict):
        container.update(addition)
        return None
    return (
        f"its collection is a {type(addition).__name__}, "
        f"but the first page's is a {type(container).__name__}"
    )


def _follow_pagination(
    session: requests.Session,
    method: str,
    request_kwargs: dict,
    accepted: set[int],
    endpoint: dict,
    spec: dict,
    document: object,
    meta: dict,
    headers: object,
    read_bytes: int,
    scrub: Callable[[str], str],
    debug: bool,
) -> tuple[str | None, bool]:
    """Read the remaining pages and merge them into ``document``.

    Returns ``(error, retryable)`` like ``_attempt`` does, and records what it
    took in ``meta``: how many pages were read, how large the merged collection
    is, and - where a next page existed but was not followed - why not.

    A page that cannot be read fails the whole endpoint rather than being
    dropped. Half a collection looks exactly like a shrinking one: 'unhealthy
    nodes: 0' because page 2 timed out is the failure mode this whole feature
    exists to remove, so it is reported as an error the endpoint's services can
    show instead.

    Every page reuses the first page's session, headers, authentication and
    timeout - the same request, at a different URL - so a cursor that only the
    API understands needs no configuration here.
    """
    source = _next_source(spec)
    items = _items_path(spec)
    if source is None or items is None:
        # The rule asked for pagination without saying where the next page or the
        # collection is. Reported rather than ignored: the services would
        # otherwise describe one page while the rule claims to follow them all.
        return "Pagination is configured without a next-page link or a collection path", False
    found, container = _resolve_path(document, items)
    if not found:
        return f"Pagination: the response has no collection at '{items or '$'}'", False
    if not isinstance(container, (list, dict)):
        return (
            f"Pagination: '{items or '$'}' is a {type(container).__name__}, "
            "not a collection that pages can be appended to"
        ), False

    max_pages, max_elements = _pagination_limits(spec)
    url = endpoint.get("url", "")
    origin = urlsplit(url).netloc.lower()
    seen = {url}
    current_url = url
    page_document = document
    page_headers = headers
    pages = 1
    total_bytes = read_bytes
    stopped: str | None = None

    while (candidate := _next_candidate(source, page_document, page_headers)) is not None:
        # The caps are only consulted once a further page actually exists, so an
        # API with exactly as many pages as the limit is read whole and reported
        # as complete - a truncation note has to mean something was left behind.
        if pages >= max_pages:
            stopped = f"the page limit ({max_pages}) was reached"
            break
        if max_elements is not None and _container_len(container) >= max_elements:
            stopped = f"the element limit ({max_elements}) was reached"
            break
        next_url, reason = _next_target(candidate, current_url, origin, seen)
        if next_url is None:
            stopped = reason
            break
        page = pages + 1
        _debug(debug, f"  page {page}: {method} {scrub(next_url)}")
        try:
            response = session.request(method, next_url, **request_kwargs)
            with response:
                status = response.status_code
                if not (200 <= status < 300 or status in accepted):
                    reason_phrase = getattr(response, "reason", "") or ""
                    return (
                        f"Page {page} failed: HTTP {status}"
                        f"{f' {reason_phrase}' if reason_phrase else ''}",
                        _retryable_status(status),
                    )
                # The 50 MiB ceiling is a budget for the whole endpoint, not per
                # page: a hundred pages of half a megabyte are just as able to
                # exhaust the monitoring host as one huge body.
                raw = _read_capped(response, _MAX_RESPONSE_BYTES - total_bytes)
                page_headers = dict(response.headers)
        except _ResponseTooLarge:
            return (
                f"Pagination exceeds the {_MAX_RESPONSE_BYTES}-byte limit at page {page}",
                False,
            )
        except requests.exceptions.RequestException as exc:
            return f"Page {page} failed: Request failed: {scrub(str(exc))}", True
        try:
            page_document = json.loads(raw)
        except ValueError as exc:
            return f"Page {page} is not valid JSON: {exc}", False
        found, addition = _resolve_path(page_document, items)
        if not found:
            return f"Page {page} has no collection at '{items or '$'}'", False
        if (mismatch := _merge_page(container, addition)) is not None:
            return f"Page {page} cannot be merged: {mismatch}", False
        total_bytes += len(raw)
        pages = page
        seen.add(next_url)
        current_url = next_url

    if stopped is not None:
        _debug(debug, f"  pagination stopped after {pages} pages: {stopped}")
    meta["pages"] = pages
    meta["elements"] = _container_len(container)
    meta["pagination_stopped"] = stopped
    # What the endpoint really read, across every page: the response size is a
    # cost the operator is watching, and one page of it is not the cost.
    meta["size"] = total_bytes
    return None, False


def _fetch(
    endpoint: dict, secret: str | None, debug: bool = False
) -> tuple[object | None, str | None, dict]:
    """Fetch one endpoint, retrying a transient failure: ``(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.

    With a retry policy configured, a failure a repeat could fix is tried again
    after a doubling wait. ``meta["attempts"]`` counts what it took, so an API
    that only answers on the second attempt is reported rather than quietly
    smoothed over - and the response time stays the successful attempt's, not
    the sum, so the metric keeps meaning what it says.
    """
    meta: dict[str, object] = {
        "status": None,
        "elapsed": None,
        "size": None,
        "final_url": None,
        "cert_expiry": None,
        "from_cache": False,
        "cache_age": None,
        "attempts": 1,
        # Response headers, so an extraction can read one (see _HEADER_PREFIX).
        # Cached alongside the body: replaying the body with someone else's
        # headers would describe a response that never existed.
        "headers": {},
    }
    # 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.
    # No request is made, so there is nothing to retry either.
    if (ttl := _cache_ttl(endpoint)) is not None and (hit := _cache_read(endpoint, ttl, secret)):
        cached_body, cached_meta = hit
        _debug(debug, f"  served from cache ({cached_meta['cache_age']:.0f}s old)")
        try:
            document = json.loads(cached_body)
        except ValueError as exc:
            # Cached something unparseable: fall through and fetch fresh.
            _debug(debug, f"  cached body is not valid JSON ({exc}), fetching")
        else:
            # The body being served IS the cached one, so that is what gets
            # reported - rebuilt here rather than replayed from the cache file,
            # so it follows the settings in force now.
            _store_report(endpoint, cached_meta, cached_body, secret)
            return document, None, cached_meta

    retries, backoff = _retry_policy(endpoint)
    slept = 0.0
    for attempt in range(retries + 1):
        meta["attempts"] = attempt + 1
        document, error, retryable = _attempt_with_token_refresh(
            endpoint, secret, meta, ttl, debug
        )
        if error is None or not retryable or attempt == retries:
            return document, error, meta
        delay = min(backoff * (2**attempt), _MAX_RETRY_SLEEP - slept)
        if delay <= 0 and slept >= _MAX_RETRY_SLEEP:
            _debug(debug, "  retry budget exhausted, giving up")
            return document, error, meta
        _debug(debug, f"  {error} - retrying in {max(delay, 0.0):.1f}s")
        if delay > 0:
            time.sleep(delay)
            slept += delay
    # Unreachable: the loop always returns (range is never empty).
    return None, "Request failed", meta


def _attempt_with_token_refresh(
    endpoint: dict, secret: str | None, meta: dict, ttl: float | None, debug: bool
) -> tuple[object | None, str | None, bool]:
    """One request, redone once with a fresh token if a cached one was rejected.

    A provider can invalidate an access token before its stated expiry (a
    rotated client secret, a revoked grant), which reaches us as a 401 the
    endpoint would otherwise report until the cached token finally expired.

    Only a token that came from the CACHE earns the second attempt. A token
    minted seconds ago and rejected means the credentials or the scope are
    wrong, and asking again would just double every check's requests forever.
    """
    oauth2 = _oauth2_spec(endpoint)
    used_cached_token = oauth2 is not None and _cached_token(oauth2, secret) is not None
    document, error, retryable = _attempt(endpoint, secret, meta, ttl, debug)
    if error is None or not used_cached_token or meta.get("status") != 401:
        return document, error, retryable
    _debug(debug, "  HTTP 401 with a cached token - discarding it and retrying once")
    return _attempt(endpoint, secret, meta, ttl, debug, force_new_token=True)


def _attempt(
    endpoint: dict,
    secret: str | None,
    meta: dict,
    ttl: float | None,
    debug: bool,
    force_new_token: bool = False,
) -> tuple[object | None, str | None, bool]:
    """One request: ``(document, error, retryable)``, updating ``meta`` in place.

    ``retryable`` says whether repeating this exact request could plausibly
    succeed. A parse failure, an oversized body and a 4xx are deterministic, so
    they are reported as final however many retries are configured.
    """
    started = time.monotonic()

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

    access_token = None
    if (oauth2 := _oauth2_spec(endpoint)) is not None:
        if force_new_token:
            _forget_token(oauth2, secret)
        try:
            access_token = _access_token(endpoint, oauth2, secret, debug)
        except _TokenError as exc:
            # No token, no request. Retryable: the provider being briefly
            # unreachable is the same class of blip as the API being so.
            _debug(debug, f"  {exc}")
            _timed()
            return None, str(exc), True

    session, headers = _build_session(endpoint, secret, access_token)
    params = _auth_params(endpoint, secret)
    # Only the query-parameter mode can carry the key into a URL, so that is the
    # only mode whose reported text needs scrubbing.
    scrub = (lambda text: _redact_secret(text, secret)) if params else (lambda text: text)
    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, _auth_header_name(endpoint)).items():
            _debug(debug, f"  header {name}: {value}")
        if session.auth is not None:
            _debug(debug, "  basic auth: <redacted>")
        if params:
            _debug(debug, f"  query parameter {next(iter(params))}: {_REDACTED}")
        if body is not None:
            _debug(debug, f"  body: {body}")
    # Collected rather than passed inline: a paginated endpoint sends the same
    # request again at the next page's URL, and every setting - authentication,
    # headers, TLS, timeout - has to be the first page's.
    request_kwargs: dict[str, object] = {
        "params": params,
        "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
    }
    try:
        response = session.request(method, endpoint["url"], **request_kwargs)
        with response:
            status = response.status_code
            meta["status"] = status
            meta["final_url"] = scrub(response.url)
            meta["headers"] = dict(response.headers)
            # 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 = scrub(response.headers.get("Location", "?"))
                _debug(debug, f"  HTTP {response.status_code} redirect to {location} (blocked)")
                _timed()
                # A blocked redirect is a property of the request, not a hiccup.
                return (
                    None,
                    f"Unexpected {status} redirect to {location} (redirects disabled)",
                    False,
                )
            # 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 ""
                # The body of a rejected response is where an API explains
                # itself, and this is the case an operator most needs to see -
                # so read it, but only when the rule asked for it (otherwise the
                # status alone is the answer and the body is never touched).
                if (report := _report_spec(endpoint)) is not None:
                    try:
                        partial, more = _read_reportable(response, _reported_limit(report))
                    except requests.exceptions.RequestException as exc:
                        # Reporting the body is a diagnostic nicety; failing to
                        # read it must not change what the endpoint reports.
                        _debug(debug, f"  could not read the error body: {exc}")
                    else:
                        _store_report(endpoint, meta, partial, secret, complete=not more)
                _timed()
                return (
                    None,
                    f"HTTP {status}{f' {reason}' if reason else ''}",
                    _retryable_status(status),
                )
            raw = _read_capped(response, _MAX_RESPONSE_BYTES)
    except _ResponseTooLarge as exc:
        # The API really does answer with that much; asking again changes nothing.
        _debug(debug, f"  {exc}")
        _timed()
        return None, str(exc), False
    except requests.exceptions.RequestException as exc:
        # Connection reset, DNS hiccup, TLS handshake, timeout: the transient
        # class this whole feature exists for. The message quotes the URL it was
        # trying to reach - query string and all - so it is scrubbed first.
        failure = scrub(str(exc))
        _debug(debug, f"  request failed: {failure}")
        _timed()
        return None, f"Request failed: {failure}", True

    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}")

    _store_report(endpoint, meta, raw, secret)
    try:
        document = json.loads(raw)
    except ValueError as exc:
        # The endpoint answered, just not with JSON - a configuration problem,
        # not a blip.
        _timed()
        return None, f"Response is not valid JSON: {exc}", False
    cacheable = raw
    if (pagination := _pagination_spec(endpoint)) is not None:
        error, retryable = _follow_pagination(
            session,
            method,
            request_kwargs,
            accepted,
            endpoint,
            pagination,
            document,
            meta,
            dict(response.headers),
            len(raw),
            scrub,
            debug,
        )
        if error is not None:
            _timed()
            return None, error, retryable
        if meta.get("pages", 1) > 1:
            # The MERGED document is what the extractions saw, so it is what a
            # cache hit has to replay - caching the first page would serve a
            # collection that shrinks for the length of the TTL. Re-serialized
            # rather than concatenated: the pages were merged as JSON.
            cacheable = json.dumps(document).encode("utf-8")
    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, cacheable, timed, secret)
    return document, None, False


_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 _inventory_spec(spec: dict, row_key: str | None) -> dict | None:
    """This field's place in the inventory tree, resolved for one result.

    The attribute name defaults to the JSON path's last segment - the same rule
    the service labels use - so the common case needs no extra typing. ``row_key``
    is the '[*]' element's label, which becomes the table row's key column;
    without a wildcard the value is a plain attribute of the node.
    """
    inventory = spec.get("inventory")
    if not isinstance(inventory, dict) or not inventory.get("node"):
        return None
    return {
        "node": str(inventory["node"]).strip(),
        "key": (inventory.get("key") or "").strip() or _label_key_from_path(spec["path"]),
        "row_key": row_key,
        "keep_service": bool(inventory.get("keep_service")),
    }


def _result(
    spec: dict,
    service: str,
    found: bool,
    value: object,
    error: str,
    url: str,
    labels: list[dict] | None = None,
    host: str | None = None,
    summary_fields: dict[str, str] | None = None,
    row_key: str | None = None,
    calc_other: object = None,
    host_labels: dict[str, str] | None = None,
    label: str | None = None,
    context: 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,
        # This value's own name WITHIN that service, for a field reported in a
        # shared service; None for the ordinary one-field-one-service case. The
        # check turns it into the line's label, so several fields can share a
        # service and still say which is which.
        "label": label,
        # 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,
        # Host labels for THAT piggyback host, resolved from this element. Also
        # internal routing, stripped with 'host': they belong to the host, not to
        # the service, so they are merged per host in _split_by_host.
        "host_labels": host_labels or {},
        "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"),
        # The second operand of a two-path transform, already resolved in this
        # result's own scope - only the agent has the document and the current
        # '[*]' element, and 'other' means "this element's total", not the
        # document's. None when no second path is configured or it did not
        # resolve; the check then reports the expression as failed rather than
        # quietly computing with a stand-in.
        "calc_other": calc_other,
        "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 [],
        # Extra summary text: the template as configured, plus the values its
        # '{path}' placeholders resolved to in THIS element's scope. Split that
        # way because only the agent can resolve a path and only the check knows
        # how the value itself is rendered.
        "summary": spec.get("summary") or None,
        "summary_fields": summary_fields or {},
        # Where this value goes in the HW/SW inventory tree, if anywhere. None
        # for the ordinary "this is a service" case.
        "inventory": _inventory_spec(spec, row_key),
        # The JSON this value was read from, for the endpoints that asked for it.
        # It travels WITH the result rather than being looked up from the
        # endpoint record by the check, because a piggybacked result lands in a
        # different section, which has no endpoint record to look anything up in.
        "context": context,
    }


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]] = []
    for label, (_default, element) in zip(_element_labels(pairs, label_path), pairs):
        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


# A path naming a RESPONSE HEADER rather than a field of the body, e.g.
# '@header.X-RateLimit-Remaining'. A prefix rather than a second form field: the
# extraction's 'path' stays the one required "where the value comes from", so no
# existing rule needs migrating and the Explorer's value_raw shape is unchanged.
# A body key literally called '@header' is shadowed by this - documented in the
# ruleset help, and no API in practice has one.
_HEADER_PREFIX = "@header."


def _resolve_header(headers: object, path: str) -> tuple[bool, object]:
    """Resolve a '@header.<name>' path against the response headers.

    HTTP field names are case-insensitive (RFC 9110), and the name reaches us as
    whatever the operator typed, so the lookup is case-folded rather than exact.
    """
    name = path[len(_HEADER_PREFIX) :].strip()
    if not isinstance(headers, dict) or not name:
        return False, None
    wanted = name.lower()
    for header, value in headers.items():
        if isinstance(header, str) and header.lower() == wanted:
            return True, value
    return False, None


def _service_prefix(endpoint: dict) -> str:
    """The endpoint name to put in front of this endpoint's service names, or ''.

    Opt-in per endpoint. Two endpoints extracting the same fields otherwise
    produce two identically named services ('JSON Status' twice, the second
    disambiguated to 'JSON Status (2)'), and nothing in the name says which
    application each belongs to.

    Deliberately no fall back to the URL: a URL in a service description travels
    into notifications, availability reports and the metric paths on disk, where
    it is both unreadable and - for a query string - a secret leak. Without a
    configured name the option is simply a no-op (the ruleset rejects that
    combination, so it can only reach here from a hand-written rule).
    """
    if not endpoint.get("service_prefix"):
        return ""
    name = endpoint.get("name")
    return name.strip() if isinstance(name, str) and name.strip() else ""


def _prefixed(prefix: str, service: str) -> str:
    """``service`` behind the endpoint prefix, if there is one."""
    return f"{prefix} {service}" if prefix else service


def _shared_service(spec: dict) -> str | None:
    """The shared service this field reports into, or None for one of its own.

    A field that names one puts its value into that service as a LINE rather than
    becoming a service itself: the check yields one result per line and Checkmk's
    own aggregation makes the service's state the worst of them. The field's own
    'service' name then names the line instead of the service.
    """
    group = spec.get("group")
    return group.strip() if isinstance(group, str) and group.strip() else None


def _extract(
    document: object,
    extractions: list[dict],
    url: str,
    headers: object = None,
    prefix: str = "",
    context: dict | None = None,
    secret: str | None = None,
) -> list[dict]:
    results = []
    for spec in extractions:
        label_specs = spec.get("labels") or []
        # Resolved in the same scope as the service labels: the current '[*]'
        # element, or the document root where there is no element.
        summary = spec.get("summary")
        # This field's service name, endpoint prefix included. Every branch below
        # names its service from this, so a prefixed endpoint cannot end up with
        # some of its services prefixed and some not.
        #
        # Naming a shared service moves the names one step along: that service is
        # what the field reports into, and the field's own name becomes the label
        # of its line inside it.
        shared = _shared_service(spec)
        base = _prefixed(prefix, shared if shared is not None else spec["service"])
        line = spec["service"] if shared is not None else None
        # A header is a single scalar off the response, so none of the body-only
        # machinery (wildcards, aggregation, filters) applies to it.
        if spec["path"].strip().startswith(_HEADER_PREFIX):
            found, value = _resolve_header(headers, spec["path"].strip())
            results.append(
                _result(
                    spec,
                    base,
                    found,
                    value,
                    "header not in response",
                    url,
                    _resolve_labels(label_specs, document),
                    summary_fields=_resolve_summary(summary, document),
                    label=line,
                    context=_context_text(context, document, spec["path"], secret=secret),
                )
            )
            continue

        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,
                    base,
                    found,
                    value,
                    error,
                    url,
                    labels,
                    summary_fields=_resolve_summary(summary, document),
                    calc_other=_resolve_calc_other(spec, document),
                    label=line,
                    context=_context_text(context, document, spec["path"], secret=secret),
                )
            )
            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,
                    base,
                    found,
                    value,
                    error,
                    url,
                    labels,
                    summary_fields=_resolve_summary(summary, document),
                    calc_other=_resolve_calc_other(spec, document),
                    label=line,
                    context=_context_text(context, document, spec["path"], secret=secret),
                )
            )
            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 index, (label_segments, found, value, error, element) in enumerate(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)
            # Only meaningful once the element IS its own host; on the polling
            # host they would silently become labels of the polling host, which
            # is emphatically not what "label the host I created" asked for.
            element_host_labels = (
                _resolve_host_labels(spec.get("piggyback_labels") or [], element)
                if host is not None
                else {}
            )
            label = " / ".join(label_segments)
            # With a shared service the expansion fans out into LINES of that one
            # service, not into services - so the element label lands on the line
            # and the service name stays put.
            if line is not None:
                service = base
                element_line = f"{line} {label}" if label else line
            elif host is not None:
                service = base
                element_line = None
            else:
                service = f"{base} {label}" if label else base
                element_line = None
            labels = _resolve_labels(label_specs, element)
            results.append(
                _result(
                    spec,
                    service,
                    found,
                    value,
                    error,
                    url,
                    labels,
                    host,
                    summary_fields=_resolve_summary(summary, element),
                    calc_other=_resolve_calc_other(spec, element),
                    host_labels=element_host_labels,
                    # An element whose label field resolved to an empty string
                    # still needs a row of its own: falling back to None would
                    # quietly turn it into a plain attribute of a node that is
                    # otherwise a table, and several such elements would then
                    # overwrite each other under one key.
                    row_key=label or str(index),
                    label=element_line,
                    context=_context_text(context, document, spec["path"], element, secret),
                )
            )
    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 = _strip_root(path).replace("[*]", "")  # a wildcard is not part of the key
    key = None
    for match in _PATH_TOKEN.finditer(cleaned):
        if (token := _token_key(match)) is not None:
            key = token
    return key or cleaned


# One '{path}' placeholder of a summary template. Braces cannot nest, so the
# body is anything but a brace; the ruleset rejects the shapes this would miss.
_SUMMARY_PLACEHOLDER = re.compile(r"\{([^{}]+)\}")

# How much of a collection to describe in a summary. Dumping the JSON of a
# 200-element array into a service summary - which travels into notifications -
# helps nobody, so a non-scalar is reported by its size instead.
def _summary_value(value: object) -> str:
    if isinstance(value, list):
        return f"[{len(value)} items]"
    if isinstance(value, dict):
        return f"{{{len(value)} keys}}"
    if value is None:
        return "null"
    text = _label_value(value)
    return text if text is not None else str(value)


def _resolve_calc_other(spec: dict, element: object) -> object:
    """Resolve the transform's second path within ``element``, or ``None``.

    Resolved in the same scope as the service's labels and summary - the current
    '[*]' element, or the document root where there is no element - so
    'value / other * 100' over 'disks[*].used' with 'total' compares each disk
    against ITS OWN total rather than against some other element's.
    """
    path = spec.get("calc_path")
    if not isinstance(path, str) or not path.strip():
        return None
    found, value = _resolve_path(element, path.strip())
    return value if found else None


def _resolve_summary(template: object, element: object) -> dict[str, str]:
    """Resolve a summary template's '{path}' placeholders against ``element``.

    Returns ``{path: rendered}`` for the placeholders that resolved, keyed by the
    path exactly as written in the template. Unresolvable ones are simply absent:
    the check marks them '(n/a)', so a typo stays visible instead of quietly
    rendering as nothing.
    """
    if not isinstance(template, str) or not template.strip():
        return {}
    fields: dict[str, str] = {}
    for match in _SUMMARY_PLACEHOLDER.finditer(template):
        path = match.group(1).strip()
        if not path or path in fields:
            continue
        found, value = _resolve_path(element, path)
        if found:
            fields[path] = _summary_value(value)
    return fields


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. ``document`` is the scope the paths are read from: the
    response root for an endpoint's own host labels, one '[*]' element for the
    labels of the piggyback host that element becomes.

    Three 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).
    * a literal ``value``: the value is typed in the rule rather than read from
      the response, and a wildcard path then collapses to ONE label - keyed once,
      not once per element - emitted as soon as an element survives the filter.
      This is the classification case: "if any of these elements matches, tag the
      host". A path is optional here; without one only the filter is evaluated.

    An optional ``filter`` (the same predicate an extraction uses) decides which
    elements produce a label. It is resolved in the same scope as the label path:
    per element for a wildcard path, against ``document`` itself otherwise - so a
    plain-path label can be made conditional too ("only in production").

    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")
        filt = spec.get("filter")
        raw_literal = spec.get("value")
        literal = raw_literal if isinstance(raw_literal, str) and raw_literal else None
        base_key = spec.get("key") or (_label_key_from_path(path) if path else "")
        if not base_key:
            continue  # nothing to key the label by (a path-less spec without a key)
        if not path:
            # Keyed and valued entirely by the rule: the only thing read from the
            # response is the filter's verdict on this scope.
            if literal is not None and _matches_filter(document, filt):
                labels[base_key] = literal
            continue
        value_field = spec.get("value_field") or ""
        segments = _split_wildcards(path)
        if len(segments) == 1:
            found, value = _resolve_path(document, path)
            if not found or not _matches_filter(document, filt):
                continue
            text = literal if literal is not None else _label_value(value)
            if text is not None:
                labels[base_key] = text
            continue
        # Wildcard: one unique label per element - or, with a literal value, one
        # label for the whole collection, as soon as an element matches.
        for label_segments, _found, _value, _error, element in _expand_wildcards(
            document, segments, None
        ):
            # The sentinel means the collection itself was missing, which is not
            # an element and must not be labelled as one.
            if element is _NO_ELEMENT or not _matches_filter(element, filt):
                continue
            if literal is not None:
                labels[base_key] = literal
                break
            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 default_label, element in 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),
        # Whether this endpoint prefixes its field service names. The check needs
        # it to name the endpoint's OWN service: an endpoint whose fields read
        # 'JSON <name> Status' wants 'JSON <name> API' rather than 'JSON API
        # <name>', or the one service describing the request sorts away from
        # every service it describes.
        "prefixed": bool(_service_prefix(endpoint)),
        "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"),
        # How many attempts the request took (1 = no retry was needed). Reported
        # so a retry policy cannot silently hide an API that is degrading.
        "attempts": meta.get("attempts", 1),
        # Pagination: how many pages were read, how many elements the merged
        # collection holds, and - when a further page existed but was not
        # followed - why not. 1 page and no reason is what an endpoint that does
        # not paginate reports, which is also the default.
        "pages": meta.get("pages", 1),
        "elements": meta.get("elements"),
        "pagination_stopped": meta.get("pagination_stopped"),
        # The raw response, for the endpoints that opted in: the body as it came
        # off the wire (capped, secret-stripped) and the response headers
        # (credential-bearing ones masked). None/absent means "not requested",
        # which is the default.
        "body": meta.get("body"),
        "body_truncated": bool(meta.get("body_truncated")),
        "body_size": meta.get("body_size"),
        "headers": meta.get("headers_reported"),
    }


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}")
    prefix = _service_prefix(endpoint)

    def _fail(error: str, meta: dict | None = None) -> tuple[list[dict], dict, dict]:
        # Prefixed exactly like the success path: a service that renamed itself
        # while the endpoint was down would go stale and its replacement would
        # be undiscovered, which is precisely when the monitoring is needed.
        results = [
            _result(
                {"path": spec.get("path", "?")},
                _prefixed(prefix, _shared_service(spec) or spec.get("service", "?")),
                False,
                None,
                error,
                url,
                label=spec.get("service", "?") if _shared_service(spec) else None,
            )
            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,
            meta.get("headers"),
            prefix,
            _context_spec(endpoint),
            secret,
        )
        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]], dict[str, dict[str, str]]]:
    """Partition results into the polling host's own and per-piggyback-host.

    ``host`` and ``host_labels`` are internal routing hints, not part of the
    section format, so they are 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.

    The third return value is the host labels each piggyback host earned, merged
    across every extraction that placed a service on it (later wins per key) -
    they describe the HOST, so two fields of the same element must not each
    write their own section-level label map.
    """
    own: list[dict] = []
    piggybacked: dict[str, list[dict]] = {}
    labels: dict[str, dict[str, str]] = {}
    for result in results:
        host = result.pop("host", None)
        host_labels = result.pop("host_labels", None) or {}
        if host is None:
            own.append(result)
            continue
        piggybacked.setdefault(host, []).append(result)
        if host_labels:
            labels.setdefault(host, {}).update(host_labels)
    return own, piggybacked, labels


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, piggyback_labels = _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": piggyback_labels.get(host, {})}
            )
            + "\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())
