#!/usr/bin/env python3
# Copyright (C) 2026 Christian Wirtz <doc@snowheaven.de>
# License: GPL-2.0-only, see LICENSE in the repository root.
"""Checkmk special agent: AWS certificates (ACM and IAM server certificates).

Discovers all TLS certificates of an AWS account in ACM (per region) and IAM
(global) and writes them as one JSON object per line. Designed to need as few
API calls as possible - see CONCEPT.md for the reasoning:

    acm:ListCertificates          one call per region (+1 per 1000 certificates)
    iam:ListServerCertificates    one global call, if IAM is enabled
    acm:DescribeCertificate       only for certificates pending renewal or with a
                                  failed/pending status, cached, with a budget
    tag:GetResources              one call per region, only with a tag filter

Within the query interval (default 1 day) every source is answered from a
local cache without any API call. The check plug-in computes the remaining
validity from NotAfter on every check run, so cached data never distorts it.
Opt-in regions that are not enabled in the account are recognized and skipped
instead of being reported as errors.

Sections produced:
    <<<aws_acm_certificates:sep(0)>>>          one ACM certificate per line
    <<<aws_iam_server_certificates:sep(0)>>>   one IAM server certificate per line
    <<<aws_acm_errors:sep(0)>>>                one API error per line
    <<<aws_acm_agent:sep(0)>>>                 run statistics and data age

Only stable, public APIs are used (boto3 as shipped with every Checkmk site),
no internal classes of Checkmk's own agent_aws.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import sys
import tempfile
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable

AGENT_VERSION = "0.1.3"

# Without an explicit keyTypes filter, ListCertificates returns RSA_2048 only and
# ECDSA certificates are silently missing (concept, rule 2).
KEY_TYPES = (
    "RSA_1024",
    "RSA_2048",
    "RSA_3072",
    "RSA_4096",
    "EC_prime256v1",
    "EC_secp384r1",
    "EC_secp521r1",
)
LIST_PAGE_SIZE = 1000
TAGGING_PAGE_SIZE = 100
IAM_PAGE_SIZE = 1000
# Hard cap so a broken NextToken loop can never run away.
MAX_PAGES = 100

DEFAULT_QUERY_INTERVAL = 24 * 3600
MIN_QUERY_INTERVAL = 15 * 60
MAX_QUERY_INTERVAL = 24 * 3600
DEFAULT_DESCRIBE_BUDGET = 20
DEFAULT_DETAILS_WINDOW_DAYS = 60
# A describe result is reused while ARN, NotAfter and Status are unchanged, but
# never longer than this, so a stuck renewal status cannot stay stale forever.
DESCRIBE_MAX_AGE = 24 * 3600
# After a failed list call, wait at least this long before calling again. Without
# it a failing region would be retried on every agent run, i.e. every minute,
# which only makes throttling worse.
ERROR_RETRY_INTERVAL = 15 * 60
# Kept small: an unreachable regional endpoint costs MAX_ATTEMPTS * CONNECT_TIMEOUT
# seconds of agent runtime, and there can be dozens of regions.
MAX_ATTEMPTS = 3
CONNECT_TIMEOUT = 5
READ_TIMEOUT = 30
CACHE_FORMAT = 1

DESCRIBE_STATUSES = frozenset({"FAILED", "VALIDATION_TIMED_OUT", "PENDING_VALIDATION"})
THROTTLING_CODES = frozenset({
    "Throttling",
    "ThrottlingException",
    "ThrottledException",
    "RequestThrottled",
    "RequestThrottledException",
    "TooManyRequestsException",
    "RequestLimitExceeded",
    "SlowDown",
})
ACCESS_DENIED_CODES = frozenset({
    "AccessDenied",
    "AccessDeniedException",
    "UnauthorizedOperation",
})
# What AWS answers in an opt-in region (e.g. af-south-1, eu-central-2) that is not
# enabled for the account. The same codes also mean "invalid credentials", so a
# region only counts as not enabled if another region worked in the same run.
REGION_DISABLED_CODES = frozenset({
    "UnrecognizedClientException",
    "InvalidClientTokenId",
})
REGION_RE = re.compile(r"^[a-z]{2}(-[a-z]+)+-\d+$")
IAM_SOURCE = "IAM"


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def iso(value: Any) -> str | None:
    """datetime or epoch -> ISO 8601 in UTC; anything else -> None."""
    if isinstance(value, datetime):
        if value.tzinfo is None:
            value = value.replace(tzinfo=timezone.utc)
        return value.astimezone(timezone.utc).isoformat()
    if isinstance(value, (int, float)):
        return datetime.fromtimestamp(value, tz=timezone.utc).isoformat()
    return None


def parse_iso(value: Any) -> float | None:
    if not value:
        return None
    try:
        return datetime.fromisoformat(str(value).replace("Z", "+00:00")).timestamp()
    except ValueError:
        return None


def api_name(method: str) -> str:
    """'list_certificates' -> 'ListCertificates', as AWS documents the operation."""
    return "".join(part.title() for part in method.split("_"))


def host_offset(hostname: str, interval: int) -> int:
    """A fixed, host-specific offset into the query interval.

    Refreshes happen at slot boundaries shifted by this offset, so several hosts
    or sites querying the same account do not all hit AWS in the same minute.
    """
    digest = hashlib.sha256(hostname.encode("utf-8")).hexdigest()
    return int(digest[:8], 16) % max(interval, 1)


def is_fresh(fetched_at: float | None, now: float, interval: int, offset: int) -> bool:
    """True while `fetched_at` and `now` lie in the same (shifted) query slot."""
    if fetched_at is None or fetched_at > now:
        return False
    return (fetched_at - offset) // interval == (now - offset) // interval


def safe_name(value: str) -> str:
    return re.sub(r"[^A-Za-z0-9._-]", "_", value) or "_"


# ---------------------------------------------------------------------------
# Cache
# ---------------------------------------------------------------------------


def default_cache_dir(hostname: str) -> Path:
    base = os.environ.get("OMD_ROOT")
    root = (Path(base) / "tmp" / "check_mk" / "special_agents" if base
            else Path(tempfile.gettempdir()) / "check_mk_special_agents")
    return root / "agent_aws_acm" / safe_name(hostname)


class CacheStore:
    """One small JSON file per cache entry, written atomically."""

    def __init__(self, directory: Path) -> None:
        self.directory = directory

    def _path(self, name: str) -> Path:
        return self.directory / f"{safe_name(name)}.json"

    def load(self, name: str) -> dict[str, Any]:
        try:
            data = json.loads(self._path(name).read_text(encoding="utf-8"))
        except (OSError, ValueError):
            return {}
        if not isinstance(data, dict) or data.get("format") != CACHE_FORMAT:
            return {}
        return data

    def save(self, name: str, data: dict[str, Any]) -> None:
        data = {**data, "format": CACHE_FORMAT}
        try:
            self.directory.mkdir(parents=True, exist_ok=True)
            fd, tmp = tempfile.mkstemp(dir=self.directory, prefix=".tmp-", suffix=".json")
            with os.fdopen(fd, "w", encoding="utf-8") as handle:
                json.dump(data, handle)
            os.replace(tmp, self._path(name))
        except OSError as exc:
            # A read-only cache must not cost the monitoring data of this run.
            sys.stderr.write(f"Cannot write cache {self._path(name)}: {exc}\n")


# ---------------------------------------------------------------------------
# AWS access
# ---------------------------------------------------------------------------


class AwsError(Exception):
    def __init__(self, operation: str, code: str, message: str) -> None:
        super().__init__(f"{operation}: {code}: {message}")
        self.operation = operation
        self.code = code
        self.message = message

    @property
    def access_denied(self) -> bool:
        return self.code in ACCESS_DENIED_CODES

    @property
    def throttled(self) -> bool:
        return self.code in THROTTLING_CODES


@dataclass
class Stats:
    calls: int = 0
    retries: int = 0
    throttles: int = 0
    describe_calls: int = 0


class Aws:
    """Lazily creates the boto3 session, so a run served from cache costs nothing,
    not even the sts:AssumeRole call."""

    def __init__(self, args: argparse.Namespace, stats: Stats) -> None:
        self.args = args
        self.stats = stats
        self._session: Any = None
        self._session_error: AwsError | None = None
        self._clients: dict[tuple[str, str], Any] = {}

    def _config(self) -> Any:
        from botocore.config import Config

        kwargs: dict[str, Any] = {
            "retries": {"mode": "adaptive", "max_attempts": MAX_ATTEMPTS},
            "connect_timeout": CONNECT_TIMEOUT,
            "read_timeout": READ_TIMEOUT,
            "user_agent_extra": f"checkmk-aws-acm/{AGENT_VERSION}",
        }
        if self.args.proxy:
            kwargs["proxies"] = {"http": self.args.proxy, "https": self.args.proxy}
        return Config(**kwargs)

    def home_region(self) -> str:
        """Region for the global services (STS, IAM), matching the partition."""
        first = self.args.regions[0] if self.args.regions else "us-east-1"
        if first.startswith("cn-"):
            return "cn-north-1"
        if first.startswith("us-gov-"):
            return "us-gov-west-1"
        return "us-east-1"

    def session(self) -> Any:
        if self._session is not None:
            return self._session
        if self._session_error is not None:
            raise self._session_error
        import boto3

        if self.args.access_key_id:
            base = boto3.Session(aws_access_key_id=self.args.access_key_id,
                                 aws_secret_access_key=self.args.secret_access_key)
        else:
            base = boto3.Session()
        if not self.args.assume_role_arn:
            self._session = base
            return base
        sts = self._register(base.client("sts", region_name=self.home_region(),
                                         config=self._config()))
        kwargs = {"RoleArn": self.args.assume_role_arn,
                  "RoleSessionName": "checkmk-aws-acm"}
        if self.args.external_id:
            kwargs["ExternalId"] = self.args.external_id
        try:
            creds = self.call(sts, "assume_role", **kwargs)["Credentials"]
        except AwsError as exc:
            # Remember the failure: every region would otherwise try again.
            self._session_error = exc
            raise
        self._session = boto3.Session(aws_access_key_id=creds["AccessKeyId"],
                                      aws_secret_access_key=creds["SecretAccessKey"],
                                      aws_session_token=creds["SessionToken"])
        return self._session

    def _register(self, client: Any) -> Any:
        client.meta.events.register("needs-retry", self._on_needs_retry)
        return client

    def _on_needs_retry(self, response: Any = None, **_kwargs: Any) -> None:
        # Called after every attempt; only counts, never influences the retry.
        if isinstance(response, tuple) and len(response) == 2:
            parsed = response[1] or {}
            if (parsed.get("Error") or {}).get("Code") in THROTTLING_CODES:
                self.stats.throttles += 1

    def client(self, service: str, region: str) -> Any:
        key = (service, region)
        if key not in self._clients:
            self._clients[key] = self._register(
                self.session().client(service, region_name=region, config=self._config()))
        return self._clients[key]

    def call(self, client: Any, method: str, **kwargs: Any) -> dict[str, Any]:
        from botocore.exceptions import BotoCoreError, ClientError

        self.stats.calls += 1
        try:
            response = getattr(client, method)(**kwargs)
        except ClientError as exc:
            meta = exc.response.get("ResponseMetadata") or {}
            self.stats.retries += int(meta.get("RetryAttempts") or 0)
            error = exc.response.get("Error") or {}
            raise AwsError(api_name(method), str(error.get("Code") or "ClientError"),
                           str(error.get("Message") or exc)) from exc
        except BotoCoreError as exc:
            raise AwsError(api_name(method), type(exc).__name__, str(exc)) from exc
        meta = response.get("ResponseMetadata") or {}
        self.stats.retries += int(meta.get("RetryAttempts") or 0)
        return response


# ---------------------------------------------------------------------------
# Fetching
# ---------------------------------------------------------------------------


def summarize_acm(region: str, summary: dict[str, Any]) -> dict[str, Any]:
    """Keep only the fields the check plugin evaluates (concept, 'Agent output')."""
    cert: dict[str, Any] = {"Region": region}
    for key in ("CertificateArn", "DomainName", "Type", "Status", "InUse",
                "RenewalEligibility", "KeyAlgorithm"):
        if summary.get(key) is not None:
            cert[key] = summary[key]
    if "DomainName" not in cert:
        # Seen on old imported certificates without a common name.
        names = summary.get("SubjectAlternativeNameSummaries") or []
        if names:
            cert["DomainName"] = names[0]
    not_after = iso(summary.get("NotAfter"))
    if not_after:
        cert["NotAfter"] = not_after
    return cert


def list_acm_certificates(aws: Aws, region: str) -> list[dict[str, Any]]:
    client = aws.client("acm", region)
    certs: list[dict[str, Any]] = []
    kwargs: dict[str, Any] = {"MaxItems": LIST_PAGE_SIZE,
                              "Includes": {"keyTypes": list(KEY_TYPES)}}
    for _ in range(MAX_PAGES):
        response = aws.call(client, "list_certificates", **kwargs)
        certs.extend(summarize_acm(region, s)
                     for s in response.get("CertificateSummaryList") or [])
        token = response.get("NextToken")
        if not token:
            break
        kwargs["NextToken"] = token
    return certs


def tag_filters(tags: list[str]) -> list[dict[str, Any]]:
    """['env=prod', 'env=stage', 'team'] -> filters ANDed across keys, ORed per key."""
    grouped: dict[str, list[str]] = {}
    for tag in tags:
        key, sep, value = tag.partition("=")
        values = grouped.setdefault(key.strip(), [])
        if sep and value.strip() and value.strip() not in values:
            values.append(value.strip())
    filters = []
    for key, values in grouped.items():
        entry: dict[str, Any] = {"Key": key}
        if values:
            entry["Values"] = values
        filters.append(entry)
    return filters


def tagged_arns(aws: Aws, region: str, tags: list[str]) -> set[str]:
    client = aws.client("resourcegroupstaggingapi", region)
    arns: set[str] = set()
    kwargs: dict[str, Any] = {"TagFilters": tag_filters(tags),
                              "ResourceTypeFilters": ["acm:certificate"],
                              "ResourcesPerPage": TAGGING_PAGE_SIZE}
    for _ in range(MAX_PAGES):
        response = aws.call(client, "get_resources", **kwargs)
        arns.update(m["ResourceARN"] for m in response.get("ResourceTagMappingList") or []
                    if m.get("ResourceARN"))
        token = response.get("PaginationToken")
        if not token:
            break
        kwargs["PaginationToken"] = token
    return arns


def fetch_region(aws: Aws, region: str, tags: list[str]) -> list[dict[str, Any]]:
    certs = list_acm_certificates(aws, region)
    if tags:
        wanted = tagged_arns(aws, region, tags)
        certs = [c for c in certs if c.get("CertificateArn") in wanted]
    return certs


def fetch_iam(aws: Aws) -> list[dict[str, Any]]:
    client = aws.client("iam", aws.home_region())
    certs: list[dict[str, Any]] = []
    kwargs: dict[str, Any] = {"MaxItems": IAM_PAGE_SIZE}
    for _ in range(MAX_PAGES):
        response = aws.call(client, "list_server_certificates", **kwargs)
        for meta in response.get("ServerCertificateMetadataList") or []:
            cert = {"ServerCertificateName": meta.get("ServerCertificateName"),
                    "Arn": meta.get("Arn")}
            expiration = iso(meta.get("Expiration"))
            if expiration:
                cert["Expiration"] = expiration
            certs.append(cert)
        if not response.get("IsTruncated") or not response.get("Marker"):
            break
        kwargs["Marker"] = response["Marker"]
    return certs


@dataclass
class SourceResult:
    name: str
    items: list[dict[str, Any]] = field(default_factory=list)
    fetched_at: float | None = None
    refreshed: bool = False
    stale: bool = False
    disabled: bool = False
    error: dict[str, Any] | None = None


def cached_fetch(store: CacheStore, cache_name: str, source: str, fingerprint: str,
                 fetch: Callable[[], list[dict[str, Any]]], now: float, interval: int,
                 offset: int) -> SourceResult:
    """Serve from cache within the query slot, otherwise fetch.

    On failure the last good data is kept and marked stale (concept, rate limits,
    measure 4), and the source is not retried for ERROR_RETRY_INTERVAL.
    """
    cache = store.load(cache_name)
    if cache.get("fingerprint") != fingerprint:
        # Credentials, role or tag filter changed: old data must not leak through.
        cache = {}
    result = SourceResult(name=source, items=list(cache.get("items") or []),
                          fetched_at=cache.get("fetched_at"),
                          disabled=bool(cache.get("disabled")))
    last_error = cache.get("last_error")
    last_attempt = float(cache.get("last_attempt_at") or 0)

    if not last_error and is_fresh(result.fetched_at, now, interval, offset):
        return result
    if last_error and 0 <= now - last_attempt < min(ERROR_RETRY_INTERVAL, interval):
        result.stale = True
        result.error = last_error
        return result

    try:
        items = fetch()
    except AwsError as exc:
        error = {"Source": source, "Operation": exc.operation, "Code": exc.code,
                 "Message": exc.message, "At": iso(now)}
        store.save(cache_name, {**cache, "fingerprint": fingerprint,
                                "last_error": error, "last_attempt_at": now})
        result.stale = True
        result.error = error
        return result

    store.save(cache_name, {"fingerprint": fingerprint, "fetched_at": now, "items": items,
                            "last_error": None, "last_attempt_at": now})
    return SourceResult(name=source, items=items, fetched_at=now, refreshed=True)


# ---------------------------------------------------------------------------
# DescribeCertificate (concept, rule 3)
# ---------------------------------------------------------------------------


def needs_details(cert: dict[str, Any], now: float, window_days: int) -> bool:
    if cert.get("Status") in DESCRIBE_STATUSES:
        return True
    if cert.get("RenewalEligibility") != "ELIGIBLE":
        return False
    not_after = parse_iso(cert.get("NotAfter"))
    return not_after is not None and not_after - now < window_days * 86400


def details_key(cert: dict[str, Any]) -> str:
    return "|".join(str(cert.get(k) or "") for k in ("CertificateArn", "NotAfter", "Status"))


def summarize_details(description: dict[str, Any], now: float) -> dict[str, Any]:
    details: dict[str, Any] = {}
    renewal = description.get("RenewalSummary") or {}
    if renewal.get("RenewalStatus"):
        details["RenewalStatus"] = renewal["RenewalStatus"]
    if renewal.get("RenewalStatusReason"):
        details["RenewalStatusReason"] = renewal["RenewalStatusReason"]
    if description.get("FailureReason"):
        details["FailureReason"] = description["FailureReason"]
    options = renewal.get("DomainValidationOptions") or description.get(
        "DomainValidationOptions") or []
    validation = [
        {"DomainName": o.get("DomainName"), "ValidationStatus": o.get("ValidationStatus"),
         "ValidationMethod": o.get("ValidationMethod")}
        for o in options if o.get("ValidationStatus") not in (None, "SUCCESS")
    ]
    if validation:
        details["Validation"] = validation[:10]
    details["DetailsFetchedAt"] = iso(now)
    return details


@dataclass
class DetailsStats:
    enabled: bool = False
    candidates: int = 0
    calls: int = 0
    pending: int = 0
    access_denied: bool = False


def attach_details(aws: Aws, store: CacheStore, certs: list[dict[str, Any]], now: float,
                   budget: int, window_days: int, interval: int,
                   errors: list[dict[str, Any]]) -> DetailsStats:
    stats = DetailsStats(enabled=True)
    cache = store.load("describe")
    entries: dict[str, Any] = cache.get("entries") or {}
    denied_at = cache.get("access_denied_at")
    if denied_at is not None and 0 <= now - float(denied_at) < interval:
        stats.access_denied = True

    candidates = [c for c in certs if c.get("CertificateArn")
                  and needs_details(c, now, window_days)]
    stats.candidates = len(candidates)
    # Failed issuance first, then the certificate expiring soonest.
    candidates.sort(key=lambda c: (c.get("Status") not in DESCRIBE_STATUSES,
                                   c.get("NotAfter") or ""))
    stop = False
    for cert in candidates:
        arn = cert["CertificateArn"]
        entry = entries.get(arn) or {}
        same_cert = entry.get("key") == details_key(cert)
        if same_cert and 0 <= now - float(entry.get("fetched_at") or 0) < DESCRIBE_MAX_AGE:
            cert["Details"] = entry["details"]
            continue
        if stats.access_denied:
            cert["Details"] = {"Error": "AccessDenied: no permission for "
                                        "acm:DescribeCertificate"}
            continue
        if stop or stats.calls >= budget:
            stats.pending += 1
            if same_cert:
                # Older details are better than none; their age is visible.
                cert["Details"] = entry["details"]
            continue
        try:
            client = aws.client("acm", cert["Region"])
            stats.calls += 1
            aws.stats.describe_calls += 1
            response = aws.call(client, "describe_certificate", CertificateArn=arn)
        except AwsError as exc:
            if exc.access_denied:
                stats.access_denied = True
                cache["access_denied_at"] = now
                cert["Details"] = {"Error": f"{exc.code}: no permission for "
                                            "acm:DescribeCertificate"}
                continue
            errors.append({"Source": cert["Region"], "Operation": exc.operation,
                           "Code": exc.code, "Message": exc.message, "At": iso(now)})
            if exc.code != "ResourceNotFoundException":
                stop = True  # throttled or unreachable: leave the rest for the next run
            if same_cert:
                cert["Details"] = entry["details"]
            continue
        details = summarize_details(response.get("Certificate") or {}, now)
        entries[arn] = {"key": details_key(cert), "fetched_at": now, "details": details}
        cert["Details"] = details

    # Drop entries of certificates that no longer exist or no longer qualify.
    live = {c["CertificateArn"] for c in candidates}
    cache["entries"] = {arn: e for arn, e in entries.items() if arn in live}
    if not stats.access_denied:
        cache.pop("access_denied_at", None)
    store.save("describe", cache)
    return stats


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------


def fingerprint(args: argparse.Namespace, *extra: Any) -> str:
    identity = [args.access_key_id or "default-chain", args.assume_role_arn or "",
                args.external_id or "", *extra]
    return hashlib.sha256(json.dumps(identity).encode("utf-8")).hexdigest()


def mark_disabled_regions(store: CacheStore, regions: list[tuple[SourceResult, str, str]],
                          now: float) -> None:
    """Turn "invalid token" errors into "region not enabled", if another region worked.

    Opt-in regions that are not enabled for the account answer every call with
    UnrecognizedClientException. That is not an error worth a WARN, and they need
    no retry before the next query slot, so they are cached like an empty region.
    If no region works at all, the credentials are the problem and the errors stay.
    """
    if not any(r.fetched_at is not None and not r.error and not r.disabled
               for r, _name, _fp in regions):
        return
    for result, cache_name, region_fingerprint in regions:
        if not result.error or result.error.get("Code") not in REGION_DISABLED_CODES:
            continue
        store.save(cache_name, {"fingerprint": region_fingerprint, "fetched_at": now,
                                "items": [], "disabled": True, "last_error": None,
                                "last_attempt_at": now})
        result.items = []
        result.fetched_at = now
        result.stale = False
        result.error = None
        result.disabled = True


def write_section(name: str, lines: list[dict[str, Any]]) -> None:
    sys.stdout.write(f"<<<{name}:sep(0)>>>\n")
    for line in lines:
        sys.stdout.write(json.dumps(line, sort_keys=True) + "\n")


def run(args: argparse.Namespace, aws: Aws | None = None, now: float | None = None) -> int:
    now = time.time() if now is None else now
    stats = Stats()
    aws = aws or Aws(args, stats)
    aws.stats = stats
    store = CacheStore(Path(args.cache_dir) if args.cache_dir
                       else default_cache_dir(args.hostname))
    interval = args.query_interval
    offset = host_offset(args.hostname, interval)

    sources: list[SourceResult] = []
    certs: list[dict[str, Any]] = []
    regions: list[tuple[SourceResult, str, str]] = []
    for region in args.regions:
        cache_name = f"list_{region}"
        region_fingerprint = fingerprint(args, region, sorted(args.tags))
        result = cached_fetch(store, cache_name, region, region_fingerprint,
                              lambda r=region: fetch_region(aws, r, args.tags),
                              now, interval, offset)
        regions.append((result, cache_name, region_fingerprint))
    mark_disabled_regions(store, regions, now)
    for result, _cache_name, _fingerprint in regions:
        sources.append(result)
        certs.extend(result.items)

    iam_certs: list[dict[str, Any]] = []
    if args.iam:
        result = cached_fetch(store, "iam", IAM_SOURCE, fingerprint(args, IAM_SOURCE),
                              lambda: fetch_iam(aws), now, interval, offset)
        sources.append(result)
        iam_certs = result.items

    errors = [s.error for s in sources if s.error]
    details = DetailsStats()
    if args.details:
        details = attach_details(aws, store, certs, now, args.describe_budget,
                                 args.details_window, interval, errors)

    write_section("aws_acm_certificates", certs)
    if args.iam:
        write_section("aws_iam_server_certificates", iam_certs)
    write_section("aws_acm_errors", errors)
    write_section("aws_acm_agent", [{
        "AgentVersion": AGENT_VERSION,
        "QueryInterval": interval,
        "Now": iso(now),
        "Sources": {
            s.name: {"FetchedAt": iso(s.fetched_at) if s.fetched_at else None,
                     "Refreshed": s.refreshed, "Stale": s.stale, "Disabled": s.disabled,
                     "Items": len(s.items)}
            for s in sources
        },
        "ApiCalls": stats.calls,
        "Retries": stats.retries,
        "Throttles": stats.throttles,
        "Details": {"Enabled": details.enabled, "Candidates": details.candidates,
                    "Calls": details.calls, "Pending": details.pending,
                    "AccessDenied": details.access_denied},
    }])
    return 0


def _region(value: str) -> str:
    if not REGION_RE.match(value):
        raise argparse.ArgumentTypeError(f"not an AWS region: {value!r}")
    return value


def _interval(value: str) -> int:
    seconds = int(value)
    if not MIN_QUERY_INTERVAL <= seconds <= MAX_QUERY_INTERVAL:
        raise argparse.ArgumentTypeError(
            f"query interval must be between {MIN_QUERY_INTERVAL} and "
            f"{MAX_QUERY_INTERVAL} seconds")
    return seconds


def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0],
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--hostname", required=True,
                        help="Checkmk host name; used for the cache directory and to "
                             "spread queries of several hosts over the interval")
    parser.add_argument("--access-key-id", help="AWS access key ID; without it the default "
                        "credential chain (environment, instance profile) is used")
    parser.add_argument("--secret-access-key", help="AWS secret access key")
    parser.add_argument("--assume-role-arn", help="IAM role to assume before querying")
    parser.add_argument("--external-id", help="external ID for the assume role call")
    parser.add_argument("--region", dest="regions", action="append", type=_region,
                        default=[], help="region to query, may be repeated")
    parser.add_argument("--no-iam", dest="iam", action="store_false",
                        help="do not query IAM server certificates")
    parser.add_argument("--query-interval", type=_interval, default=DEFAULT_QUERY_INTERVAL,
                        help="seconds between AWS queries; within it the cache answers "
                             f"(default {DEFAULT_QUERY_INTERVAL})")
    parser.add_argument("--no-details", dest="details", action="store_false",
                        help="never call acm:DescribeCertificate")
    parser.add_argument("--describe-budget", type=int, default=DEFAULT_DESCRIBE_BUDGET,
                        help="maximum DescribeCertificate calls per run "
                             f"(default {DEFAULT_DESCRIBE_BUDGET})")
    parser.add_argument("--details-window", type=int, default=DEFAULT_DETAILS_WINDOW_DAYS,
                        help="fetch renewal details for eligible certificates expiring "
                             f"within this many days (default {DEFAULT_DETAILS_WINDOW_DAYS})")
    parser.add_argument("--tag", dest="tags", action="append", default=[],
                        help="only ACM certificates with this tag, KEY or KEY=VALUE; "
                             "may be repeated (different keys: AND, same key: OR)")
    proxy = parser.add_mutually_exclusive_group()
    proxy.add_argument("--proxy", help="proxy URL, e.g. http://user:pass@proxy:3128")
    proxy.add_argument("--no-proxy", action="store_true",
                       help="ignore proxy settings from the environment")
    parser.add_argument("--cache-dir", help="override the cache directory")
    parser.add_argument("--debug", action="store_true", help="raise Python exceptions")
    args = parser.parse_args(argv)
    if args.access_key_id and not args.secret_access_key:
        parser.error("--access-key-id needs --secret-access-key")
    if args.describe_budget < 0 or args.details_window < 0:
        parser.error("--describe-budget and --details-window must not be negative")
    if not args.regions and not args.iam:
        parser.error("nothing to query: give at least one --region or enable IAM")
    return args


def main(argv: list[str] | None = None) -> int:
    args = parse_arguments(argv)
    if args.no_proxy:
        for name in ("HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy"):
            os.environ.pop(name, None)
    try:
        return run(args)
    except Exception as exc:  # pylint: disable=broad-except
        if args.debug:
            raise
        sys.stderr.write(f"agent_aws_acm: {type(exc).__name__}: {exc}\n")
        return 1


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