#!/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"]}, ...]}
"""

import argparse
import json
import re
import sys
from collections.abc import Sequence
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import requests
from cmk.utils import password_store as _legacy_pwstore

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

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

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


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

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


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


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

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


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

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


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


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


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


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

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


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

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


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


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


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

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


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

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


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

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


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

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


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


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

    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:
        return json.loads(raw), None
    except ValueError as exc:
        return None, f"Response is not valid JSON: {exc}"


_WILDCARD = "[*]"


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 _result(
    spec: dict,
    service: str,
    found: bool,
    value: object,
    error: str,
    url: str,
    labels: list[dict] | None = None,
) -> dict:
    # "Count elements": monitor the length of the list/object at the path instead
    # of the value itself. A scalar has no length to count, so that is surfaced as
    # a not-found misconfiguration rather than counting characters of a string.
    if found and spec.get("count"):
        if isinstance(value, (list, dict)):
            elements = value if isinstance(value, list) else list(value.values())
            filt = spec.get("filter")
            if isinstance(filt, dict) and filt.get("path"):
                # Count only the elements matching the filter (e.g. how many
                # nodes are NOT 'ok'), not the whole collection.
                elements = [element for element in elements if _matches_filter(element, filt)]
            value = len(elements)
        else:
            found = False
            error = "value at path is not a list or object (cannot count elements)"
    # 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,
        "path": spec["path"],
        "url": url,
        "found": found,
        "value": value if found else None,
        "error": None if found else error,
        "levels_upper": spec.get("levels_upper"),
        "levels_lower": spec.get("levels_lower"),
        "match": spec.get("match"),
        "calc": spec.get("calc"),
        "unit": spec.get("unit"),
        "labels": labels or [],
    }


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

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


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

    ``segments`` is the output of :func:`_split_wildcards`: one entry per
    wildcard level plus the trailing value path. Returns one
    ``(label_segments, found, value, error, element)`` tuple per leaf - i.e. per
    element of the cartesian product of all wildcard levels. ``label_segments``
    carries one label per wildcard level (joined into the composite service name
    by the caller); ``label_path`` is resolved relative to each level's element,
    with the array index (or object key) as the fallback. ``element`` is the leaf
    element node (``None`` 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", None)]

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


def _extract(document: object, extractions: list[dict], url: str) -> list[dict]:
    results = []
    for spec in extractions:
        label_specs = spec.get("labels") or []
        segments = _split_wildcards(spec["path"])
        if len(segments) == 1:
            found, value = _resolve_path(document, spec["path"])
            labels = _resolve_labels(label_specs, document)
            results.append(
                _result(
                    spec, spec["service"], found, value, "path not found in response", url, labels
                )
            )
            continue

        # One or more array wildcards: one service per cartesian-product
        # element, labelled by a ' / '-joined composite (one segment per level).
        # An optional filter keeps only the elements whose sub-field matches
        # (e.g. only the nodes that are NOT 'ok'); a not-found leaf is kept so
        # the "array or object not found" error still surfaces.
        filt = spec.get("filter")
        for label_segments, found, value, error, element in _expand_wildcards(
            document, segments, spec.get("label_path")
        ):
            if found and isinstance(filt, dict) and filt.get("path"):
                if not _matches_filter(element, filt):
                    continue
            label = " / ".join(label_segments)
            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))
    return results


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

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


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

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


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

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


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

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

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

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


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

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


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

    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.
    """
    url = endpoint.get("url", "?")
    debug = getattr(args, "debug", False)
    _debug(debug, f"endpoint {index}: {url}")

    def _fail(error: str) -> tuple[list[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)], {})

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


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] = {}
    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 in per_endpoint:
                results.extend(endpoint_results)
                host_labels.update(endpoint_host_labels)  # later endpoints win per key

    payload = {"results": results, "host_labels": host_labels}
    sys.stdout.write("<<<json_api:sep(0)>>>\n")
    sys.stdout.write(json.dumps(payload) + "\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())
