#!/usr/bin/env python3
# Copyright (C) 2026 Christian Wirtz <doc@snowheaven.de>
# License: GPL-2.0-only, see LICENSE in the repository root.
"""Special agent: MKP Updates

Lives at:
    local/lib/python3/cmk_addons/plugins/mkp_updates/libexec/agent_mkp_updates

Enumerates all MKP extension packages installed on this Checkmk site (via
``mkp list --json``) and, for every package that has a "Download-URL" set,
tries to determine the latest available version:

  * github.com and self-hosted GitHub Enterprise -> GitHub REST API
  * gitlab.com and self-hosted GitLab             -> GitLab REST API
  * Gitea / Forgejo instances                     -> Gitea REST API
  * anything else (including an offline/local git
    server with no REST API at all)               -> ``git ls-remote --tags``

Which of these applies to an unknown host is decided by briefly probing it
at runtime (see ``_probe_provider``) -- there is no fixed list of hosts to
maintain.

For any package that still has no usable version after the above (no
Download-URL, or the lookup failed/found nothing), the Checkmk Exchange
catalog (https://exchange.checkmk.com/api/packages/all -- undocumented and
unofficial, cached locally and re-fetched at most once a day, regardless
of how often this agent itself runs) is checked as a fallback --
see ``fetch_exchange_catalog()`` and
``mkp_updates_core.match_exchange_version()``. Disable with --no-exchange.

The output is a single JSON document in one Checkmk agent section, parsed
by the ``mkp_updates`` check plug-in.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path
from urllib.parse import quote

import requests

sys.path.insert(0, str(Path(__file__).resolve().parent))
import mkp_updates_core as core  # noqa: E402

AGENT_SECTION_NAME = "mkp_updates"
USER_AGENT = (
    "checkmk-mkp-updates-special-agent/1.0 "
    "(+https://gitlab.com/checkmk1/checkmk_mkp-updates)"
)

# Cap on how long we're willing to spend probing an *unknown* host to guess
# its provider, independent of the (potentially much larger) overall
# --timeout, so one slow/dead host can't eat the whole agent run just for
# the auto-detection step.
_PROBE_TIMEOUT_CAP = 5.0

# Undocumented but real (verified by hand): a single request returns *all* Exchange
# packages, no auth, no pagination. See mkp_updates_core's "Checkmk Exchange fallback
# matching" section for the response shape and why matching is done the way it is.
_EXCHANGE_API_URL = "https://exchange.checkmk.com/api/packages/all"


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def parse_arguments(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser("agent_mkp_updates", description=__doc__)
    parser.add_argument(
        "--timeout",
        type=float,
        default=10.0,
        help="Per-request network timeout in seconds (default: 10).",
    )
    parser.add_argument(
        "--include-prereleases",
        action="store_true",
        help="Consider pre-release tags/releases as the latest version.",
    )
    parser.add_argument(
        "--insecure",
        action="store_true",
        help="Disable TLS certificate verification for HTTPS requests and git operations.",
    )
    parser.add_argument(
        "--no-exchange",
        action="store_true",
        help=(
            "Disable the Checkmk Exchange fallback lookup (used for packages whose "
            "Download-URL is missing or didn't resolve a version). Fetches "
            f"{_EXCHANGE_API_URL} once per run when enabled (the default)."
        ),
    )
    parser.add_argument(
        "--show-status",
        action="append",
        default=[],
        dest="show_statuses",
        metavar="STATUS",
        help=(
            "Only include packages with this status (e.g. 'update_available', "
            "'error') in the service details; may be given multiple times. Every "
            "package is still checked and counted regardless -- this only limits "
            "what the check plug-in renders in the details, not the summary/state. "
            "Not meant to be set by hand -- this is what the special agent's "
            "server_side_calls plugin passes. If omitted entirely (e.g. manual CLI "
            "use), nothing is filtered."
        ),
    )
    parser.add_argument(
        "--credential-host",
        action="append",
        default=[],
        dest="credential_hosts",
        metavar="HOST_PATTERN",
        help=(
            "Regular expression matching a git host's name. Paired, by position, with "
            "the --credential-token-id given alongside it. May be given multiple times; "
            "the first matching pattern wins. Not meant to be set by hand -- this is "
            "what the special agent's server_side_calls plugin passes."
        ),
    )
    parser.add_argument(
        "--credential-token-id",
        action="append",
        default=[],
        dest="credential_token_ids",
        metavar="PASSWORD_STORE_REFERENCE",
        help=(
            "Checkmk password store reference (\"<id>:<path>\") for the token belonging "
            "to the --credential-host given alongside it, resolved via "
            "cmk.password_store.v1_unstable.dereference_secret(). Not meant to be set by "
            "hand -- this is what the special agent's server_side_calls plugin passes, "
            "precisely so the real token never ends up on this process's command line "
            "(visible via e.g. `ps`)."
        ),
    )
    parser.add_argument(
        "--mkp-json",
        type=Path,
        default=None,
        help=argparse.SUPPRESS,  # debugging aid: read `mkp list --json` output from a file
    )
    return parser.parse_args(argv)


def resolve_credentials(
    host_patterns: list[str], token_ids: list[str]
) -> list[tuple[re.Pattern, str]]:
    """Turn --credential-host/--credential-token-id pairs into
    (compiled host pattern, real token) tuples.

    Resolving a password store reference requires cmk.password_store, which is only
    importable when running inside a Checkmk >= 2.5.0 site's Python -- which every real
    invocation, via the special agent mechanism, always is. The import is deferred (not
    done at module load time) so the rest of this script stays importable/testable
    without a Checkmk installation, and so sites with no credentials configured at all
    never need the module to be present.
    """
    if len(host_patterns) != len(token_ids):
        raise SystemExit(
            "--credential-host and --credential-token-id must be given the same number "
            f"of times (got {len(host_patterns)} and {len(token_ids)})."
        )
    if not host_patterns:
        return []

    try:
        from cmk.password_store.v1_unstable import dereference_secret
    except ImportError as exc:
        raise SystemExit(
            "Got --credential-token-id, but the 'cmk.password_store' module "
            f"(Checkmk >= 2.5.0) is not available in this Python environment: {exc}"
        ) from exc

    credentials = []
    for host_pattern, token_id in zip(host_patterns, token_ids):
        try:
            pattern = re.compile(host_pattern)
        except re.error as exc:
            raise SystemExit(f"Invalid --credential-host pattern {host_pattern!r}: {exc}") from exc
        try:
            token = dereference_secret(token_id).reveal()
        except Exception as exc:
            raise SystemExit(
                f"Could not resolve --credential-token-id for host pattern {host_pattern!r}: {exc}"
            ) from exc
        credentials.append((pattern, token))
    return credentials


# ---------------------------------------------------------------------------
# Reading the site's installed MKPs
# ---------------------------------------------------------------------------


def _omd_root() -> Path:
    root = os.environ.get("OMD_ROOT")
    if not root:
        raise SystemExit("OMD_ROOT is not set -- this agent must run inside a Checkmk site.")
    return Path(root)


def load_local_packages(mkp_json_override: Path | None) -> tuple[list[dict], list[str]]:
    """Return (packages, errors), where each package is a dict with at
    least name/title/installed_version/author/download_url/state."""
    errors: list[str] = []

    if mkp_json_override is not None:
        try:
            raw = mkp_json_override.read_text()
        except OSError as exc:
            return [], [f"Could not read {mkp_json_override}: {exc}"]
    else:
        mkp_bin = _omd_root() / "bin" / "mkp"
        try:
            proc = subprocess.run(
                [str(mkp_bin), "list", "--json"],
                capture_output=True,
                text=True,
                timeout=60,
                check=False,
            )
        except OSError as exc:
            return [], [f"Failed to execute {mkp_bin}: {exc}"]
        if proc.returncode != 0:
            detail = proc.stderr.strip() or f"exit code {proc.returncode}"
            return [], [f"'mkp list --json' failed: {detail}"]
        raw = proc.stdout

    try:
        data = json.loads(raw)
    except json.JSONDecodeError as exc:
        return [], [f"Could not parse 'mkp list --json' output: {exc}"]

    installed_ids = {(m["name"], m["version"]) for m in data.get("installed", [])}
    inactive_ids = {(m["name"], m["version"]) for m in data.get("inactive", [])}
    local_manifests = data.get("stored", {}).get("local", [])

    # A site can have several stored versions of the same package at once - the
    # currently active one, plus old disabled/inactive leftovers that were never
    # cleaned up (mkp doesn't remove them on its own). Group by name first, so each
    # package name is only ever checked against the network once, no matter how many
    # versions of it are sitting in the package store.
    by_name: dict[str, list[dict]] = {}
    for manifest in local_manifests:
        try:
            name = manifest["name"]
            _version = manifest["version"]
        except (KeyError, TypeError):
            errors.append(f"Skipping malformed package manifest entry: {manifest!r}")
            continue
        by_name.setdefault(name, []).append(manifest)

    packages = []
    for name, manifests in by_name.items():
        manifest = core.pick_representative_manifest(manifests, installed_ids, inactive_ids)
        pid = (manifest["name"], manifest["version"])
        if pid in installed_ids:
            state = "enabled_active"
        elif pid in inactive_ids:
            state = "enabled_inactive"
        else:
            state = "disabled"
        packages.append(
            {
                "name": name,
                "title": manifest.get("title") or name,
                "installed_version": manifest["version"],
                "author": manifest.get("author") or "",
                "download_url": manifest.get("download_url") or "",
                "state": state,
            }
        )
    return packages, errors


# ---------------------------------------------------------------------------
# Provider auto-detection for hosts we don't recognize by name
# ---------------------------------------------------------------------------


def _probe_provider(host: str, session: requests.Session, timeout: float) -> str | None:
    """Best-effort classification of an unknown git host by briefly
    probing well-known API endpoints. Returns "gitlab", "gitea", "github"
    or None (meaning: fall back to plain 'git ls-remote')."""

    def _json_dict(resp: requests.Response) -> dict | None:
        try:
            body = resp.json()
        except ValueError:
            return None
        return body if isinstance(body, dict) else None

    try:
        resp = session.get(f"https://{host}/api/v4/version", timeout=timeout)
        if resp.status_code in (200, 401) and _json_dict(resp) is not None:
            return "gitlab"
    except requests.RequestException:
        pass

    try:
        resp = session.get(f"https://{host}/api/v1/version", timeout=timeout)
        if resp.status_code == 200 and _json_dict(resp) is not None:
            return "gitea"
    except requests.RequestException:
        pass

    try:
        resp = session.get(f"https://{host}/api/v3", timeout=timeout)
        if (
            resp.status_code in (200, 401, 403)
            and "X-GitHub-Request-Id" in resp.headers
            and _json_dict(resp) is not None
        ):
            return "github"
    except requests.RequestException:
        pass

    return None


# ---------------------------------------------------------------------------
# Provider-specific resolvers.
#
# Each of these returns a dict with keys "latest_version", "release_url",
# "provider_detail" and "error" (None on success). Turning that into a
# final status/message is done once, centrally, in resolve_latest_version().
# ---------------------------------------------------------------------------


def _github_api_base(api_host: str) -> str:
    return "https://api.github.com" if api_host == "github.com" else f"https://{api_host}/api/v3"


def _resolve_github(
    ref: core.RepoRef,
    session: requests.Session,
    timeout: float,
    include_prereleases: bool,
    token: str | None,
    api_host: str,
) -> dict:
    path = ref.github_style_path
    if path is None:
        return _err("Could not determine owner/repo from the Download-URL.")

    base = _github_api_base(api_host)
    headers = {"Accept": "application/vnd.github+json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"

    if not include_prereleases:
        resp = session.get(f"{base}/repos/{path}/releases/latest", headers=headers, timeout=timeout)
        if resp.status_code == 200:
            data = resp.json()
            tag = data.get("tag_name")
            if tag:
                return _ok(tag, data.get("html_url"), "releases")
        elif _is_github_rate_limited(resp):
            return _err("GitHub API rate limit exceeded (add a token in the rule to raise it).")
        elif resp.status_code not in (404,):
            return _err(f"GitHub API returned HTTP {resp.status_code} for 'releases/latest'.")

    resp = session.get(
        f"{base}/repos/{path}/releases", headers=headers, timeout=timeout, params={"per_page": 30}
    )
    if resp.status_code == 200:
        candidates = [
            r["tag_name"]
            for r in resp.json()
            if isinstance(r, dict) and r.get("tag_name") and (include_prereleases or not r.get("prerelease"))
        ]
        if candidates:
            best = max(candidates, key=core.version_sort_key)
            return _ok(best, f"https://{ref.host}/{path}/releases/tag/{best}", "releases")
    elif _is_github_rate_limited(resp):
        return _err("GitHub API rate limit exceeded (add a token in the rule to raise it).")

    resp = session.get(f"{base}/repos/{path}/tags", headers=headers, timeout=timeout, params={"per_page": 30})
    if resp.status_code == 200:
        tags = [t["name"] for t in resp.json() if isinstance(t, dict) and t.get("name")]
        if tags:
            best = max(tags, key=core.version_sort_key)
            return _ok(best, f"https://{ref.host}/{path}/releases/tag/{best}", "tags")
        return _no_releases()
    if _is_github_rate_limited(resp):
        return _err("GitHub API rate limit exceeded (add a token in the rule to raise it).")
    if resp.status_code == 404:
        return _err("Repository not found (renamed, deleted, or private without credentials).")
    return _err(f"GitHub API returned HTTP {resp.status_code}.")


def _is_github_rate_limited(resp: requests.Response) -> bool:
    return resp.status_code == 403 and resp.headers.get("X-RateLimit-Remaining") == "0"


def _resolve_gitlab(
    ref: core.RepoRef,
    session: requests.Session,
    timeout: float,
    include_prereleases: bool,
    token: str | None,
    api_host: str,
) -> dict:
    path = ref.full_path
    if path is None:
        return _err("Could not determine the project path from the Download-URL.")

    project_id = quote(path, safe="")
    base = f"https://{api_host}/api/v4"
    headers = {"PRIVATE-TOKEN": token} if token else {}

    resp = session.get(
        f"{base}/projects/{project_id}/releases", headers=headers, timeout=timeout, params={"per_page": 100}
    )
    if resp.status_code == 200:
        candidates = []
        for r in resp.json():
            if not isinstance(r, dict):
                continue
            tag = r.get("tag_name")
            if not tag:
                continue
            if not include_prereleases and r.get("upcoming_release"):
                continue
            candidates.append(tag)
        if candidates:
            best = max(candidates, key=core.version_sort_key)
            return _ok(best, f"https://{api_host}/{path}/-/releases/{best}", "releases")

    resp = session.get(
        f"{base}/projects/{project_id}/repository/tags", headers=headers, timeout=timeout, params={"per_page": 100}
    )
    if resp.status_code == 200:
        tags = [t["name"] for t in resp.json() if isinstance(t, dict) and t.get("name")]
        if tags:
            best = max(tags, key=core.version_sort_key)
            return _ok(best, f"https://{api_host}/{path}/-/tags/{best}", "tags")
        return _no_releases()
    if resp.status_code in (401, 403):
        return _err("GitLab API rejected the request (missing/invalid credentials for a private project?).")
    if resp.status_code == 404:
        return _err("Project not found.")
    return _err(f"GitLab API returned HTTP {resp.status_code}.")


def _resolve_gitea(
    ref: core.RepoRef,
    session: requests.Session,
    timeout: float,
    include_prereleases: bool,
    token: str | None,
    api_host: str,
) -> dict:
    path = ref.github_style_path
    if path is None:
        return _err("Could not determine owner/repo from the Download-URL.")

    base = f"https://{api_host}/api/v1"
    headers = {"Authorization": f"token {token}"} if token else {}

    resp = session.get(f"{base}/repos/{path}/releases", headers=headers, timeout=timeout, params={"limit": 50})
    if resp.status_code == 200:
        candidates = [
            r["tag_name"]
            for r in resp.json()
            if isinstance(r, dict) and r.get("tag_name") and (include_prereleases or not r.get("prerelease"))
        ]
        if candidates:
            best = max(candidates, key=core.version_sort_key)
            return _ok(best, f"https://{api_host}/{path}/releases/tag/{best}", "releases")

    resp = session.get(f"{base}/repos/{path}/tags", headers=headers, timeout=timeout, params={"limit": 50})
    if resp.status_code == 200:
        tags = [t["name"] for t in resp.json() if isinstance(t, dict) and t.get("name")]
        if tags:
            best = max(tags, key=core.version_sort_key)
            return _ok(best, f"https://{api_host}/{path}/releases/tag/{best}", "tags")
        return _no_releases()
    if resp.status_code == 404:
        return _err("Repository not found.")
    return _err(f"Gitea API returned HTTP {resp.status_code}.")


def _resolve_generic_git(ref: core.RepoRef, timeout: float, verify_tls: bool, token: str | None) -> dict:
    """Fallback for any host we couldn't otherwise classify -- including a
    purely offline/LAN-local git server with no REST API at all. Only
    needs the git wire protocol to be reachable, nothing else."""
    if shutil.which("git") is None:
        return _err("The 'git' command is not available on this Checkmk server.")

    url = core.inject_url_credentials(ref.raw_url, token)
    env = dict(os.environ)
    env["GIT_TERMINAL_PROMPT"] = "0"
    if not verify_tls:
        env["GIT_SSL_NO_VERIFY"] = "true"

    try:
        proc = subprocess.run(
            ["git", "ls-remote", "--tags", "--refs", url],
            capture_output=True,
            text=True,
            timeout=timeout,
            env=env,
            check=False,
        )
    except subprocess.TimeoutExpired:
        return _err(f"'git ls-remote' timed out after {timeout:.0f}s.")
    except OSError as exc:
        return _err(f"Failed to execute git: {exc}")

    if proc.returncode != 0:
        stderr_lines = proc.stderr.strip().splitlines()
        detail = stderr_lines[-1] if stderr_lines else f"exit code {proc.returncode}"
        return _err(f"'git ls-remote' failed: {detail}")

    tags = []
    for line in proc.stdout.splitlines():
        _sha, _, tag_ref = line.partition("\t")
        prefix = "refs/tags/"
        if tag_ref.startswith(prefix):
            tags.append(tag_ref[len(prefix):])

    if not tags:
        return _no_releases()

    best = max(tags, key=core.version_sort_key)
    return _ok(best, None, "git-tags")


def _ok(latest_version: str, release_url: str | None, provider_detail: str) -> dict:
    return {
        "latest_version": latest_version,
        "release_url": release_url,
        "provider_detail": provider_detail,
        "error": None,
    }


def _no_releases() -> dict:
    return {"latest_version": None, "release_url": None, "provider_detail": "", "error": None}


def _err(message: str) -> dict:
    return {"latest_version": None, "release_url": None, "provider_detail": "", "error": message}


# ---------------------------------------------------------------------------
# Checkmk Exchange fallback
# ---------------------------------------------------------------------------


# exchange.checkmk.com/api/packages/all is undocumented and unofficial -- not a
# published, rate-limited API meant for this kind of polling. This project has no
# visibility into how many sites end up running it, so request volume is kept low
# and *bounded independently of install count or check interval*, not just fast:
# the catalog is cached to disk and only re-fetched at most this often, regardless
# of how frequently the special agent itself is invoked (e.g. a site that ignores
# the check-interval advice elsewhere in this rule and polls every minute still
# only causes one real request per cache period, not one per run). Keep this
# comfortably long -- Exchange package releases happen at most a few times a day,
# nowhere near a cadence that benefits from checking more often than this.
_EXCHANGE_CACHE_MAX_AGE_SECONDS = 24 * 60 * 60  # 1 day


def _exchange_cache_path() -> Path:
    omd_root = os.environ.get("OMD_ROOT")
    if not omd_root:
        # Manual/CLI use outside a site (e.g. local testing): still cache, just
        # somewhere sensible, rather than skipping caching altogether.
        return Path(tempfile.gettempdir()) / "mkp_updates_exchange_cache.json"
    return Path(omd_root) / "var" / "check_mk" / "cache" / "mkp_updates" / "exchange_packages.json"


def _read_exchange_cache(path: Path, *, ignore_age: bool) -> list[dict] | None:
    try:
        age = time.time() - path.stat().st_mtime
    except OSError:
        return None
    if not ignore_age and age > _EXCHANGE_CACHE_MAX_AGE_SECONDS:
        return None
    try:
        data = json.loads(path.read_text())
    except (OSError, ValueError):
        return None
    return data if isinstance(data, list) else None


def _write_exchange_cache(path: Path, packages: list[dict]) -> None:
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(packages))
    except OSError:
        pass  # caching is a best-effort optimization; never let it be fatal


def fetch_exchange_catalog(session: requests.Session, timeout: float) -> list[dict] | None:
    """Return the Checkmk Exchange package catalog, from a local cache (see
    _EXCHANGE_CACHE_MAX_AGE_SECONDS above) if fresh enough, otherwise via one
    HTTP request -- which, on success, refreshes the cache for next time.

    Returns None only if there is truly nothing usable: no fresh cache, the
    request itself failed, *and* no stale cache exists to fall back to either.
    Callers treat that as "fallback unavailable this run", not a hard error --
    Exchange is a bonus source on top of the Download-URL, never the only one.
    """
    cache_path = _exchange_cache_path()

    cached = _read_exchange_cache(cache_path, ignore_age=False)
    if cached is not None:
        return cached

    try:
        resp = session.get(_EXCHANGE_API_URL, timeout=timeout)
        resp.raise_for_status()
        data = resp.json()
        packages = data.get("data", {}).get("packages") if isinstance(data, dict) else None
        if not isinstance(packages, list):
            raise ValueError("unexpected response shape")  # noqa: TRY301
    except (requests.RequestException, ValueError):
        # Request failed (or returned something we don't understand): fall back to
        # a stale cache rather than nothing, if one happens to exist.
        return _read_exchange_cache(cache_path, ignore_age=True)

    _write_exchange_cache(cache_path, packages)
    return packages


# ---------------------------------------------------------------------------
# Orchestration
# ---------------------------------------------------------------------------


def _resolve_via_download_url(
    pkg: dict,
    *,
    session: requests.Session,
    timeout: float,
    verify_tls: bool,
    include_prereleases: bool,
    credentials: list[tuple[re.Pattern, str]],
) -> dict:
    """The original (pre-Exchange) resolution: Download-URL -> provider ->
    latest version. Returns a result dict with provider/provider_detail/
    latest_version/release_url/status/message, same shape as
    resolve_latest_version()'s return value.
    """
    download_url = pkg.get("download_url") or ""
    ref = core.parse_repo_url(download_url)
    if ref is None:
        return {
            "provider": None,
            "provider_detail": "",
            "latest_version": None,
            "release_url": None,
            "status": core.STATUS_NO_SOURCE,
            "message": (
                "The package's Download-URL does not look like a repository link."
                if download_url
                else "The package's Download-URL field is empty."
            ),
        }

    token = core.pick_credential(ref.host, credentials)
    provider = core.guess_known_provider(ref.host)

    try:
        if provider is None:
            provider = _probe_provider(ref.host, session, min(timeout, _PROBE_TIMEOUT_CAP))

        if provider == "github":
            raw = _resolve_github(ref, session, timeout, include_prereleases, token, ref.host)
        elif provider == "gitlab":
            raw = _resolve_gitlab(ref, session, timeout, include_prereleases, token, ref.host)
        elif provider == "gitea":
            raw = _resolve_gitea(ref, session, timeout, include_prereleases, token, ref.host)
        else:
            provider = "generic_git"
            raw = _resolve_generic_git(ref, timeout, verify_tls, token)
    except requests.RequestException as exc:
        raw = {"latest_version": None, "release_url": None, "provider_detail": "", "error": f"Network error: {exc}"}
    except Exception as exc:  # never let one bad package abort the whole agent run
        raw = {"latest_version": None, "release_url": None, "provider_detail": "", "error": f"Unexpected error: {exc}"}

    if raw.get("error"):
        status = core.STATUS_ERROR
        message = raw["error"]
    elif raw.get("latest_version") is None:
        status = core.STATUS_NO_RELEASES
        message = "Repository has no tags or releases."
    else:
        status = core.classify_update(pkg["installed_version"], raw["latest_version"])
        message = ""

    return {
        "provider": provider,
        "provider_detail": raw.get("provider_detail", ""),
        "latest_version": raw.get("latest_version"),
        "release_url": raw.get("release_url"),
        "status": status,
        # Belt and braces: strip any "scheme://credential@" that might have ended up in
        # an error message (e.g. from git's own stderr echoing the URL it tried) before
        # this ever reaches a check's summary/details, which are visible to every user
        # who can see the service - not just admins with password store access.
        "message": core.redact_credentials(message),
    }


def resolve_latest_version(
    pkg: dict,
    *,
    session: requests.Session,
    timeout: float,
    verify_tls: bool,
    include_prereleases: bool,
    credentials: list[tuple[re.Pattern, str]],
    exchange_catalog: list[dict] | None = None,
) -> dict:
    result = _resolve_via_download_url(
        pkg,
        session=session,
        timeout=timeout,
        verify_tls=verify_tls,
        include_prereleases=include_prereleases,
        credentials=credentials,
    )
    return core.apply_exchange_fallback(
        result,
        pkg_name=pkg["name"],
        installed_version=pkg["installed_version"],
        download_url=pkg.get("download_url") or "",
        exchange_packages=exchange_catalog,
    )


def main(argv: list[str] | None = None) -> int:
    args = parse_arguments(sys.argv[1:] if argv is None else argv)
    verify_tls = not args.insecure
    credentials = resolve_credentials(args.credential_hosts, args.credential_token_ids)

    packages, agent_errors = load_local_packages(args.mkp_json)

    session = requests.Session()
    session.verify = verify_tls
    session.headers["User-Agent"] = USER_AGENT

    exchange_catalog = None
    if not args.no_exchange:
        exchange_catalog = fetch_exchange_catalog(session, args.timeout)
        if exchange_catalog is None:
            agent_errors.append(
                "Could not fetch the Checkmk Exchange package catalog "
                f"({_EXCHANGE_API_URL}) -- the Exchange fallback lookup is skipped "
                "for this run; Download-URL results are unaffected."
            )

    results = []
    for pkg in packages:
        info = resolve_latest_version(
            pkg,
            session=session,
            timeout=args.timeout,
            verify_tls=verify_tls,
            include_prereleases=args.include_prereleases,
            credentials=credentials,
            exchange_catalog=exchange_catalog,
        )
        results.append({**pkg, **info})

    payload = {
        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
        "site": os.environ.get("OMD_SITE", ""),
        "packages": results,
        "errors": agent_errors,
        # Which status groups the rule wants shown in the service details. Every
        # package above is still included regardless -- filtering by this happens
        # only in the check plug-in's details rendering, so the summary/state stay
        # based on the complete picture. None means "not restricted" (show all).
        "show_statuses": args.show_statuses or None,
    }

    print(f"<<<{AGENT_SECTION_NAME}:sep(0)>>>")
    print(json.dumps(payload))
    return 0


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