certbot-dns-netbox/certbot_dns_netbox/dns_netbox.py

"""NetBox DNS (netbox-plugin-dns) DNS-01 authenticator plugin for Certbot.

The TXT record required for the ACME dns-01 challenge is managed
exclusively through the netbox-dns record API endpoint, via the bundled
nbapi.py helper. The zone endpoint is never queried and the plugin never
creates or deletes records: an existing TXT record (created beforehand in
NetBox) is updated with the challenge token and left in place afterwards.

Connection settings are taken from the environment, exactly as nbapi.py
expects them:

* ``NETBOX_URL``: base URL of the NetBox instance
* ``NETBOX_BEARER`` or ``NETBOX_TOKEN``: NetBox API token
"""

import logging
import os
from typing import Callable, Optional

import requests

from certbot import errors
from certbot.plugins import dns_common

logger = logging.getLogger(__name__)

# netbox-dns view that contains the zone with the challenge TXT record.
search_view = "global"


class Authenticator(dns_common.DNSAuthenticator):
    """DNS-01 authenticator backed by the netbox-dns plugin."""

    description = "Obtain certificates using a DNS TXT record (in netbox-dns)."

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.client = None

    @classmethod
    def add_parser_arguments(
        cls, add: Callable[..., None], default_propagation_seconds: int = 30
    ) -> None:
        super().add_parser_arguments(
            add, default_propagation_seconds=default_propagation_seconds
        )
        add(
            "ttl",
            type=int,
            default=None,
            help="TTL for the challenge TXT record, in seconds "
            "(default: keep the record's current TTL).",
        )
        add(
            "insecure",
            action="store_true",
            help="Do not verify the TLS certificate of the NetBox instance.",
        )

    def more_info(self) -> str:
        return (
            "This plugin updates an existing DNS TXT record in netbox-dns "
            "with the token of the dns-01 challenge."
        )

    def _setup_credentials(self) -> None:
        # nbapi.py reads its connection settings from the environment and
        # terminates the process at import time when they are missing, so
        # verify them here and fail with a proper certbot error instead.
        if not os.environ.get("NETBOX_URL"):
            raise errors.PluginError(
                "Environment variable NETBOX_URL is not set (required by nbapi.py)."
            )
        if not (os.environ.get("NETBOX_BEARER") or os.environ.get("NETBOX_TOKEN")):
            raise errors.PluginError(
                "Environment variable NETBOX_BEARER or NETBOX_TOKEN must be set "
                "(required by nbapi.py)."
            )

        self.client = _NetboxDnsClient(
            ttl=self.conf("ttl"), insecure=self.conf("insecure")
        )

    def _perform(self, domain: str, validation_name: str, validation: str) -> None:
        try:
            self.client.update_txt_record(validation_name, validation)
        except errors.PluginError:
            raise
        except Exception as e:
            raise errors.PluginError(
                f"Failed to update TXT record {validation_name}: {e}"
            ) from e

    def _cleanup(self, domain: str, validation_name: str, validation: str) -> None:
        # The TXT record is managed outside of this plugin and must remain in
        # place so that it can be updated again on the next renewal; deleting
        # it would leave nothing to update.
        logger.debug(
            "Leaving TXT record %s in place (this plugin only updates records).",
            validation_name,
        )


class _NetboxDnsClient:
    """netbox-dns client built on the bundled nbapi helper.

    Only the record API endpoint is used, so the API token needs no
    permissions on zones. Records are never created or deleted: the
    challenge TXT record must already exist in NetBox and is updated in
    place.
    """

    def __init__(self, ttl: Optional[int] = None, insecure: bool = False):
        # Deferred import: nbapi.py exits at import time when the NETBOX_*
        # environment variables are missing. Authenticator._setup_credentials()
        # performs the environment checks before this client is constructed.
        from certbot_dns_netbox import nbapi

        self._nbapi = nbapi
        self._ttl = ttl
        if insecure:
            nbapi.disable_ssl_verify()

    @staticmethod
    def _fqdn(record_fqdn: str) -> str:
        # netbox-dns stores record fqdns with a trailing dot.
        return record_fqdn.rstrip(".").lower() + "."

    def _get_txt_records(self, fqdn: str) -> list:
        try:
            return (
                self._nbapi.get_dns_record(
                    {"fqdn": fqdn, "type": "TXT", "view": search_view},
                    multi_exp=True,
                )
                or []
            )
        except requests.exceptions.RequestException as e:
            raise errors.PluginError(
                f"Failed to query netbox-dns records for {fqdn}: {e}"
            ) from e

    def update_txt_record(self, record_fqdn: str, validation: str) -> None:
        """Write the challenge token into the existing TXT record."""
        fqdn = self._fqdn(record_fqdn)
        records = self._get_txt_records(fqdn)

        if any(record["value"] == validation for record in records):
            logger.debug(
                "TXT record for %s already has the expected value.", record_fqdn
            )
            return

        existing = [record for record in records if not record.get("managed")]
        if not existing:
            raise errors.PluginError(
                f"No updatable TXT record found for {record_fqdn} in view "
                f"{search_view!r}. This plugin only updates existing records: "
                "create the record once (type TXT) and rerun certbot."
            )

        patch = {"id": existing[0]["id"], "value": validation}
        if self._ttl:
            patch["ttl"] = self._ttl
        self._nbapi.modify_dns_record(patch)
        logger.info(
            "Updated TXT record %s (id %s).",
            existing[0]["fqdn"],
            existing[0]["id"],
        )