message-broker-certs

#!/usr/bin/env python3
# Copyright (C) 2024 Checkmk GmbH - License: GNU General Public License v2
# This file is part of Checkmk (https://checkmk.com). It is subject to the terms and
# conditions defined in the file COPYING, which is part of this source code package.
"""Ensure the message broker certificates are created"""

import argparse
import logging
import sys
from dataclasses import dataclass
from pathlib import Path

from cmk.utils.certs import SiteBrokerCA, SiteBrokerCertificate


@dataclass(frozen=True)
class Arguments:
    omd_root: Path
    site_name: str


def initialize_message_broker_certs(omd_root: Path, site_name: str) -> None:
    """Initialize the CA and create the certificate for use with the message broker.
    These might be replaced by the config sync later.
    """

    ca = SiteBrokerCA.create_and_persist(site_name, omd_root)
    ca.write_trusted_cas_file(omd_root)

    site_cert = SiteBrokerCertificate.create(site_name, omd_root, issuer=ca.cert_bundle)
    site_cert.persist(omd_root)


def _parse_arguments(argv: list[str]) -> Arguments:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("omd_root", help="The OMD root directory")
    parser.add_argument("site_name", help="The site name")

    args = parser.parse_args(argv[1:])
    return Arguments(
        omd_root=Path(args.omd_root),
        site_name=str(args.site_name),
    )


def main(argv: list[str]) -> int:
    logger = logging.getLogger("cmk-message-broker-certs")
    logger.addHandler(handler := logging.StreamHandler(stream=sys.stdout))
    handler.setFormatter(logging.Formatter("%(message)s"))
    logger.setLevel(logging.INFO)

    try:
        args = _parse_arguments(argv)
        initialize_message_broker_certs(args.omd_root, args.site_name)
    except Exception as e:
        logger.error("Error: %s", e)
        return 1

    return 0


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