From 91562ae538be24ed5614fdc4163c510e6f26a9b1 Mon Sep 17 00:00:00 2001 From: Sven Luthi Date: Mon, 26 Oct 2026 17:47:10 +0100 Subject: [PATCH] new file: README.md new file: certbot_dns_netbox/__init__.py new file: certbot_dns_netbox/dns_netbox.py new file: certbot_dns_netbox/nbapi.py new file: pyproject.toml --- README.md | 99 ++++++++++ certbot_dns_netbox/__init__.py | 3 + certbot_dns_netbox/dns_netbox.py | 170 ++++++++++++++++ certbot_dns_netbox/nbapi.py | 319 +++++++++++++++++++++++++++++++ pyproject.toml | 32 ++++ 5 files changed, 623 insertions(+) create mode 100644 README.md create mode 100644 certbot_dns_netbox/__init__.py create mode 100644 certbot_dns_netbox/dns_netbox.py create mode 100755 certbot_dns_netbox/nbapi.py create mode 100644 pyproject.toml diff --git a/README.md b/README.md new file mode 100644 index 0000000..1ec516d --- /dev/null +++ b/README.md @@ -0,0 +1,99 @@ +# certbot-dns-netbox + +Certbot DNS-01 authenticator plugin that updates the challenge TXT record in +[NetBox DNS](https://github.com/sys4/netbox-plugin-dns) (the `netbox-dns` +NetBox plugin) via the bundled `nbapi.py` API helper. + +## How it works + +For every domain in the certificate request certbot asks this plugin to set an +`_acme-challenge.` TXT record. The plugin works exclusively through +the netbox-dns **record API endpoint** — the zone endpoint is never queried, +and the plugin **never creates or deletes records**: + +1. looks up the existing TXT records for the FQDN (records can be filtered by + `fqdn`, `zone` and `view` by name), +2. writes the challenge token into the existing record. If the record does + not exist yet, certbot aborts with an error: create the record once + beforehand (in the NetBox UI or your tooling) and rerun, +3. leaves the record in place after validation so it can be updated again on + the next renewal. + +Because a single record holds a single token, a certificate request must not +combine a domain and its wildcard (`-d example.com -d '*.example.com'`): both +map to the same `_acme-challenge.example.com` record, and the second token +would overwrite the first. Issue those as separate certificates. + +Connection settings are taken from the environment, exactly as `nbapi.py` +expects them: + +| Variable | Purpose | +| --------------- | ------------------------------------------ | +| `NETBOX_URL` | Base URL of the NetBox instance | +| `NETBOX_BEARER` | NetBox API token (`Authorization: Bearer`) | +| `NETBOX_TOKEN` | NetBox API token (`Authorization: Token`) | +| `DEBUG` | Optional: verbose nbapi.py logging | + +One of `NETBOX_BEARER` / `NETBOX_TOKEN` is required. The token's user needs +NetBox object permissions to view and change netbox-dns records only — no +zone or view permissions are required. + +## Installation + +Install into the same Python environment as certbot: + +``` +pip install ./certbot-dns-netbox +``` + +## Usage + +``` +export NETBOX_URL=https://netbox.example.com +export NETBOX_BEARER= + +certbot certonly \ + --authenticator dns-netbox \ + --dns-netbox-propagation-seconds 60 \ + -d example.com -d '*.example.com' +``` + +### Options + +| Option | Default | Description | +| ---------------------------------- | -------------- | ------------------------------------ | +| `--dns-netbox-propagation-seconds` | `30` | Seconds to wait for DNS propagation | +| `--dns-netbox-ttl` | keep current | TTL of the challenge TXT record (s) | +| `--dns-netbox-insecure` | off | Skip TLS verification towards NetBox | + +The netbox-dns view to search is configured in code: `search_view` at the top +of `certbot_dns_netbox/dns_netbox.py`. + +### Renewal configuration + +certbot stores the authenticator in the renewal configuration, so renewals +need no extra flags: + +``` +# /etc/letsencrypt/renewal/example.com.conf +[renewalparams] +authenticator = dns-netbox +dns_netbox_ttl = 120 +``` + +When certbot runs unattended (cron or systemd timer), make sure the `NETBOX_*` +environment variables are set, e.g. via a systemd `EnvironmentFile=` or a +wrapper script. + +## Verification + +``` +certbot plugins --init dns-netbox # plugin is discovered and loads +certbot certonly --dry-run \ + --authenticator dns-netbox \ + -d sub.example.com # end-to-end against the staging ACME API +``` + +Watch the `_acme-challenge` TXT record in the `global` view: its value changes +to the challenge token during the run and remains in place afterwards. The +next renewal updates the same record again. diff --git a/certbot_dns_netbox/__init__.py b/certbot_dns_netbox/__init__.py new file mode 100644 index 0000000..df66f92 --- /dev/null +++ b/certbot_dns_netbox/__init__.py @@ -0,0 +1,3 @@ +"""NetBox DNS authenticator plugin for Certbot.""" + +__version__ = "1.0.0" diff --git a/certbot_dns_netbox/dns_netbox.py b/certbot_dns_netbox/dns_netbox.py new file mode 100644 index 0000000..616d61d --- /dev/null +++ b/certbot_dns_netbox/dns_netbox.py @@ -0,0 +1,170 @@ +"""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"], + ) diff --git a/certbot_dns_netbox/nbapi.py b/certbot_dns_netbox/nbapi.py new file mode 100755 index 0000000..63c1d2a --- /dev/null +++ b/certbot_dns_netbox/nbapi.py @@ -0,0 +1,319 @@ +#!/usr/bin/env python3 +import os +import sys +import requests +import urllib3 +import json +from typing import Union + +base_url = os.environ.get("NETBOX_URL", None) +bearer = os.environ.get("NETBOX_BEARER", None) +token = os.environ.get("NETBOX_TOKEN", None) +os.environ.pop("http_proxy", None) +os.environ.pop("https_proxy", None) + +DEBUG = bool(os.environ.get("DEBUG", False)) + +if not base_url: + print("❌ Missing required environment variable NETBOX_URL") + sys.exit(1) +if not bearer and not token: + print("❌ Missing required environment variables NETBOX_BEARER or NETBOX_TOKEN") + sys.exit(1) + + +# Clean URL +base_url = base_url.rstrip("/") + +ssl_verify = True +timeout = 60 + +# Following for legacy code. Replace with direct header var access: +headers = { + "Authorization": f"Bearer {bearer}" if bearer else f"Token {token}", + "Content-Type": "application/json", +} + +session = requests.Session() +session.headers.update(headers) + + +def _log_debug(msg: str): + if DEBUG: + print(msg) + + +def _raise_with_body(e, r, method="", path="", query=None, data=None): + """Re-raise *e* enriched with the NetBox API response body and request + context when it is an HTTPError. + + NetBox returns detailed validation errors in the response body (e.g. + {"mac_address": ["Enter a valid MAC address."]}), but requests' + raise_for_status() produces an HTTPError whose str() carries only the + status code and URL. This rebuilds the exception so str() includes + the API message and the request context (method, path, filters, data), + making failures debuggable without DEBUG mode. + Non-HTTPError exceptions (e.g. JSONDecodeError) are re-raised unchanged. + """ + if not isinstance(e, requests.exceptions.HTTPError): + raise e + + try: + body = r.json() + except (ValueError, json.JSONDecodeError): + body = (r.text or "")[:1000] + + if isinstance(body, dict): + parts = [] + for k, v in body.items(): + if isinstance(v, list): + parts.append(f"{k}: {', '.join(str(x) for x in v)}") + else: + parts.append(f"{k}: {v}") + body_str = "; ".join(parts) if parts else str(body) + else: + body_str = str(body) + + req_context = "" + if method or path: + req_context = f" | Request: {method} {path}" + if query: + req_context += f" filters={query}" + if data: + req_context += f" data={data}" + + raise requests.exceptions.HTTPError( + f"{e} | API: {body_str}{req_context}", response=r + ) + + +def disable_ssl_verify(): + global ssl_verify + ssl_verify = False + urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + + +# ============================================================================== +# ========== Get functions ===================================================== +# ============================================================================== +def _get( + path: str, filters: Union[dict, list] = None, raw_text: bool = False +) -> Union[dict, str]: + # Disable object limit to ensure all matching objects are returned. + if isinstance(filters, dict): + if "limit" not in filters: + filters["limit"] = 0 + elif isinstance(filters, list): + index = next((v for k, v in filters if k == "limit"), None) + if index is None: + filters.append(("limit", 0)) + else: + raise Exception( + f"_get() called with unsupported filters type '{type(filters)}'" + ) + + path = path.strip("/") + url = f"{base_url}/api/{path}/" + try: + r = session.get(url, params=filters or {}, verify=ssl_verify, timeout=timeout) + r.raise_for_status() + return r.text if raw_text else r.json() + except (requests.exceptions.HTTPError, json.JSONDecodeError) as e: + _raise_with_body(e, r, method="GET", path=path, query=filters) + + +def get_resource( + path: str, filters: Union[dict, list], multi_exp: bool = False +) -> Union[dict, None]: + data = _get(path, filters) + count = data["count"] + # No Object found + if count == 0: + _log_debug(f"No object found using filter {filters}") + return None + + # If multiple objects expected, return them all: + elif multi_exp: + return data["results"] + + # If single object is expected, return it without list: + elif count == 1: + return data["results"][0] + + # Single object expected, but multiple found, warn user and return nothing. + else: + _log_debug( + f"Error: Multiple objects found using filter {filters}. Refine your query!" + ) + return None + + +# ============================================================================== +# ========== Update functions ================================================== +# ============================================================================== +def _patch(path: str, obj_id: int, data: dict) -> bool: + path = path.strip("/") + url = f"{base_url}/api/{path}/{obj_id}/" + try: + r = session.patch(url, json=data, verify=ssl_verify, timeout=timeout) + r.raise_for_status() + if r.status_code in (200, 204): + return True + else: + _log_debug("Update failed:") + _log_debug(r.text) + return False + + except requests.exceptions.HTTPError as e: + _raise_with_body(e, r, method="PATCH", path=f"{path}/{obj_id}", data=data) + + +def _patch_resource(path: str, data: dict) -> Union[dict, None]: + obj_id = data.get("id", None) + # Dont execute if no ID is set + if not obj_id: + _log_debug("Error: No ID supplied when patching resource!") + return False + else: + return _patch( + path, + obj_id, + data, + ) + + +def modify_resource(path: str, data: dict) -> Union[dict, None]: + return _patch_resource(path, data) + + +# ============================================================================== +# ========== Create functions ================================================== +# ============================================================================== +def _post(path: str, data: dict) -> Union[dict, list]: + path = path.strip("/") + url = f"{base_url}/api/{path}/" + try: + r = session.post(url, json=data, verify=ssl_verify, timeout=timeout) + r.raise_for_status() + return r.json() + except (requests.exceptions.HTTPError, json.JSONDecodeError) as e: + _raise_with_body(e, r, method="POST", path=path, data=data) + + +def _post_resource(path: str, data: dict) -> Union[dict, None]: + if data.get("id"): + _log_debug("Error: ID supplied when creating resource!") + return False + else: + return _post( + path, + data, + ) + + +def create_resource(path: str, data: dict) -> Union[dict, None]: + return _post_resource(path, data) + + +# ============================================================================== +# ========== Delete functions ================================================== +# ============================================================================== +def _delete(path: str, obj_id: int) -> bool: + path = path.strip("/") + url = f"{base_url}/api/{path}/{obj_id}/" + try: + r = session.delete(url, verify=ssl_verify, timeout=timeout) + r.raise_for_status() + return True + except requests.exceptions.HTTPError as e: + _log_debug(f"HTTPError: {e}\nResponse body: {r.text}") + return False + + +def delete_resource(path: str, obj_id: int) -> bool: + return _delete(path, obj_id) + + +# ============================================================================== +# ========== Export functions ================================================== +# ============================================================================== +def export(path: str, filters: Union[dict, list], template: str) -> str: + if isinstance(filters, dict): + filters["export"] = template + elif isinstance(filters, list): + filters.append(("export", template)) + else: + raise Exception( + f"export() called with unsupported filters type '{type(filters)}'" + ) + return _get(path, filters, raw_text=True) + + +# ============================================================================== +# ========== Resource-specific wrappers (table-driven) ========================= +# ============================================================================== +# Single source of truth mapping resource name -> NetBox API path. The loop +# below generates get_/modify_/create_/delete_ convenience wrappers for every +# entry, delegating to the generic core functions above. Adding a new resource +# is a one-line change to RESOURCES. +RESOURCES = { + "device": "/dcim/devices", + "device_interface": "/dcim/interfaces", + "device_role": "/dcim/device-roles", + "inventory_item": "/dcim/inventory-items", + "site": "/dcim/sites", + "location": "/dcim/locations", + "rack": "/dcim/racks", + "cable": "/dcim/cables", + "platform": "/dcim/platforms", + "tenant": "/tenancy/tenants", + "vm": "/virtualization/virtual-machines", + "vm_interface": "/virtualization/interfaces", + "virtual_disk": "/virtualization/virtual-disks", + "ipaddress": "/ipam/ip-addresses", + "vlan": "/ipam/vlans", + "dns_view": "/plugins/netbox-dns/views/", + "dns_zone": "/plugins/netbox-dns/zones/", + "dns_record": "/plugins/netbox-dns/records/" +} + + +def _make_getter(path: str): + def _wrapper(filters: Union[dict, list], multi_exp: bool = False): + return get_resource(path, filters, multi_exp) + + return _wrapper + + +def _make_modifier(path: str): + def _wrapper(data: dict): + return _patch_resource(path, data) + + return _wrapper + + +def _make_creator(path: str): + def _wrapper(data: dict): + return _post_resource(path, data) + + return _wrapper + + +def _make_deleter(path: str): + def _wrapper(obj_id: int): + return _delete(path, obj_id) + + return _wrapper + + +for _name, _path in RESOURCES.items(): + for _verb, _factory in ( + ("get", _make_getter), + ("modify", _make_modifier), + ("create", _make_creator), + ("delete", _make_deleter), + ): + _fn = _factory(_path) + _fn.__name__ = f"{_verb}_{_name}" + _fn.__qualname__ = f"{_verb}_{_name}" + globals()[f"{_verb}_{_name}"] = _fn diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..d37fe6e --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,32 @@ +[build-system] +requires = ["setuptools>=61"] +build-backend = "setuptools.build_meta" + +[project] +name = "certbot-dns-netbox" +version = "1.0.0" +description = "NetBox DNS (netbox-plugin-dns) DNS-01 authenticator plugin for Certbot" +readme = "README.md" +requires-python = ">=3.9" +license = { text = "Apache-2.0" } +keywords = ["certbot", "acme", "letsencrypt", "netbox", "dns", "dns-01"] +classifiers = [ + "Development Status :: 5 - Production/Stable", + "Environment :: Plugins", + "Intended Audience :: System Administrators", + "Operating System :: POSIX :: Linux", + "Programming Language :: Python :: 3", + "Topic :: Internet :: WWW/HTTP", + "Topic :: Security", +] +dependencies = [ + "certbot>=2.7.0", + "requests>=2.20.0", + "urllib3>=1.26", +] + +[project.entry-points."certbot.plugins"] +dns-netbox = "certbot_dns_netbox.dns_netbox:Authenticator" + +[tool.setuptools] +packages = ["certbot_dns_netbox"]