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

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

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

The section payload looks like:

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

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

import argparse
import hashlib
import json
import math
import os
import re
import ssl
import sys
import tempfile
import time
from collections.abc import Sequence
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from urllib.parse import quote, quote_plus, 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 _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


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"),
        ],
        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.
        "meta": {
            k: v for k, v in meta.items() if k not in ("from_cache", "cache_age", "attempts")
        },
    }
    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


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:
            return json.loads(cached_body), None, cached_meta
        except ValueError as exc:
            # Cached something unparseable: fall through and fetch fresh.
            _debug(debug, f"  cached body is not valid JSON ({exc}), fetching")

    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}")
    try:
        response = session.request(
            method,
            endpoint["url"],
            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
        )
        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 ""
                _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}")

    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
    timed = _timed()
    if ttl is not None:
        # Only a response we could actually parse is worth caching; an error or a
        # non-JSON body would just be replayed for the whole TTL.
        _cache_write(endpoint, raw, timed, 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,
) -> dict:
    # Values that are not JSON scalars (dict/list) are reported as strings.
    if found and isinstance(value, (dict, list)):
        value = json.dumps(value)
    return {
        "service": service,
        # Which host this result belongs to: None is the polling host (its
        # services go into the plain section), a name means a piggyback host and
        # is stripped back out before the section is written.
        "host": host,
        # 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),
    }


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 _extract(
    document: object, extractions: list[dict], url: str, headers: object = 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")
        # 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,
                    spec["service"],
                    found,
                    value,
                    "header not in response",
                    url,
                    _resolve_labels(label_specs, document),
                    summary_fields=_resolve_summary(summary, document),
                )
            )
            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,
                    spec["service"],
                    found,
                    value,
                    error,
                    url,
                    labels,
                    summary_fields=_resolve_summary(summary, document),
                    calc_other=_resolve_calc_other(spec, document),
                )
            )
            continue

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

        # Aggregating a '[*]' path collapses the expansion into ONE service (the
        # sum/average/... over the elements) instead of fanning it out. Service
        # labels then have no single element to resolve against, so they come
        # from the response root.
        if aggregate:
            found, value, error = _aggregate_leaves(aggregate, leaves, filt)
            labels = _resolve_labels(label_specs, document)
            results.append(
                _result(
                    spec,
                    spec["service"],
                    found,
                    value,
                    error,
                    url,
                    labels,
                    summary_fields=_resolve_summary(summary, document),
                    calc_other=_resolve_calc_other(spec, document),
                )
            )
            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)
            if host is not None:
                service = spec["service"]
            else:
                service = f"{spec['service']} {label}" if label else spec["service"]
            labels = _resolve_labels(label_specs, element)
            results.append(
                _result(
                    spec,
                    service,
                    found,
                    value,
                    error,
                    url,
                    labels,
                    host,
                    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),
                )
            )
    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. Two shapes are supported:

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

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


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

    The label comes from label_path within each element, falling back to the
    default label (an array index or object key). If a label value repeats
    across elements, every occurrence of it is suffixed with its position, so
    two elements can never collapse into one service.
    """
    labels = []
    for 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),
        "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),
    }


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

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

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

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

    try:
        document, error, meta = _fetch(endpoint, secret, debug)
        if error is not None:
            return _fail(error, meta)
        results = _extract(document, endpoint.get("extractions", []), url, meta.get("headers"))
        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())
