chore(api): clean knot.py: add docs

Author: nils-wisiol

api/desecapi/knot.py

@@ -1,10 +1,11 @@
+"""Knot DNS backend helpers for catalog updates, DNSSEC, and transfers."""
+
 from functools import lru_cache
 from hashlib import sha1
 import logging
 import os
 import socket
 import select
-import threading
 import time
 
 import dns.dnssec
@@ -43,6 +44,7 @@
 
 
 def _tsig_algorithm(name):
+    """Return a dnspython TSIG algorithm constant for the configured name."""
     if not name:
         return None
     algorithm = _TSIG_ALGORITHM_MAP.get(name.lower())
@@ -53,6 +55,7 @@ def _tsig_algorithm(name):
 
 @lru_cache(maxsize=1)
 def _knot_host_ip():
+    """Resolve NSLORD_KNOT_HOST to a concrete IP address (IPv4/IPv6)."""
     host = settings.NSLORD_KNOT_HOST
     try:
         dns.inet.af_for_address(host)
@@ -75,6 +78,7 @@ def _knot_host_ip():
 
 
 def _update_keyring():
+    """Return TSIG keyring/name/algorithm tuple for dynamic updates."""
     key_name = settings.NSLORD_KNOT_UPDATE_KEY_NAME
     key_secret = settings.NSLORD_KNOT_UPDATE_KEY_SECRET
     if not key_name or not key_secret:
@@ -88,6 +92,7 @@ def _update_keyring():
 
 
 def _transfer_keyring():
+    """Return TSIG keyring/name/algorithm tuple for AXFR/IXFR transfers."""
     key_name = settings.NSLORD_KNOT_TRANSFER_KEY_NAME
     key_secret = settings.NSLORD_KNOT_TRANSFER_KEY_SECRET
     if not key_name or not key_secret:
@@ -132,15 +137,18 @@ def _send_update(update: dns.update.Update) -> None:
 
 
 def _catalog_member_subname(zone):
+    """Return catalog member label for a zone name (stable hash)."""
     zone = zone.rstrip(".") + "."
     return f"{sha1(zone.encode()).hexdigest()}.zones"
 
 
 def _catalog_record_name(zone):
+    """Return the FQDN of the catalog member PTR record for a zone."""
     return f"{_catalog_member_subname(zone)}.{settings.CATALOG_ZONE}".strip(".") + "."
 
 
 def _new_update(zone):
+    """Create a dnspython Update with configured TSIG for a zone."""
     keyring, keyname, keyalgorithm = _update_keyring()
     return dns.update.Update(
         zone,
@@ -178,6 +186,7 @@ def ensure_default_ns(name):
 
 
 def _write_bind_keypair(name, dnskey, private_key):
+    """Write BIND-style DNSKEY + private key files for Knot import."""
     import_dir = settings.NSLORD_KNOT_IMPORT_DIR
     if not import_dir or not private_key:
         return None
@@ -201,6 +210,7 @@ def _write_bind_keypair(name, dnskey, private_key):
 
 
 def _key_ready_path(name):
+    """Return the path of the Knot CSK import readiness marker file."""
     import_dir = settings.NSLORD_KNOT_IMPORT_DIR
     if not import_dir:
         return None
@@ -209,12 +219,14 @@ def _key_ready_path(name):
 
 
 def prepare_csk_key(name, *, dnskey, private_key=None):
+    """Prepare a CSK keypair for Knot import without triggering any update."""
     if not private_key:
         return
     _write_bind_keypair(name, dnskey, private_key)
 
 
 def wait_for_csk_key_ready(name):
+    """Wait for Knot to signal completion of CSK import via .ready file."""
     ready_path = _key_ready_path(name)
     if not ready_path:
         return
@@ -228,6 +240,7 @@ def wait_for_csk_key_ready(name):
 
 
 def _dnskey_present(name, dnskey):
+    """Check whether a specific DNSKEY RR appears in the zone."""
     query = dns.message.make_query(name, dns.rdatatype.DNSKEY, want_dnssec=True)
     response = dns.query.tcp(
         query,
@@ -245,6 +258,7 @@ def _dnskey_present(name, dnskey):
 
 
 def _wait_for_dnskey(name, dnskey, *, attempts: int = 20, delay_seconds: float = 0.2):
+    """Poll for the presence of a DNSKEY RR, with bounded retries."""
     for _ in range(attempts):
         if _dnskey_present(name, dnskey):
             return True
@@ -254,6 +268,7 @@ def _wait_for_dnskey(name, dnskey, *, attempts: int = 20, delay_seconds: float =
 
 
 def _dnskey_set(name):
+    """Return the set of DNSKEY RR text values in the zone."""
     query = dns.message.make_query(name, dns.rdatatype.DNSKEY, want_dnssec=True)
     response = dns.query.tcp(
         query,
@@ -273,6 +288,7 @@ def _dnskey_set(name):
 def _wait_for_dnskey_set(
     name, expected, *, attempts: int = 60, delay_seconds: float = 0.5
 ):
+    """Poll until the DNSKEY RRset equals the expected set."""
     for _ in range(attempts):
         if _dnskey_set(name) == expected:
             return True
@@ -282,6 +298,7 @@ def _wait_for_dnskey_set(
 
 
 def import_csk_key(name, *, dnskey, private_key=None):
+    """Import a CSK into Knot and optionally verify visibility."""
     if not wait_for_zone(name, attempts=60, interval_seconds=0.5):
         raise KnotException(f"Knot zone {name} not ready for updates")
     if private_key:
@@ -324,6 +341,7 @@ def import_csk_key(name, *, dnskey, private_key=None):
 
 
 def wait_for_zone(name, *, attempts=20, interval_seconds=0.5) -> bool:
+    """Poll for zone availability by querying SOA from Knot."""
     query = dns.message.make_query(name, dns.rdatatype.SOA)
     query_timeout = min(settings.NSLORD_KNOT_TIMEOUT, 1.0)
 
@@ -349,12 +367,14 @@ def wait_for_zone(name, *, attempts=20, interval_seconds=0.5) -> bool:
 
 
 def _sleep(seconds: float) -> None:
+    """Sleep without blocking signals in some environments."""
     if seconds <= 0:
         return
     select.select([], [], [], seconds)
 
 
 def delete_zone(name):
+    """Delete a zone from the catalog."""
     catalog_update = _new_update(settings.CATALOG_ZONE)
     catalog_update.delete(_catalog_record_name(name), "PTR")
     _send_update(catalog_update)
@@ -413,23 +433,7 @@ def update_rrsets(
         has_changes = True
 
     if has_changes:
-        update_done = {"error": None}
-
-        def _apply_update():
-            try:
-                _send_update(update)
-            except Exception as exc:
-                update_done["error"] = exc
-
-        thread = threading.Thread(target=_apply_update, daemon=True)
-        thread.start()
-        thread.join(timeout=settings.NSLORD_KNOT_TIMEOUT * 2)
-        if thread.is_alive():
-            if settings.DEBUG:
-                return
-            raise KnotException("Knot update timed out")
-        if update_done["error"] is not None:
-            raise update_done["error"]
+        _send_update(update)
 
 
 def import_zonefile_rrsets(name, rrsets):
@@ -463,6 +467,7 @@ def import_zonefile_rrsets(name, rrsets):
 def ensure_soa_serial_min(
     name, serial: int, *, attempts: int = 5, delay_seconds: float = 0.2
 ):
+    """Ensure SOA serial is at least the given value by issuing updates."""
     query = dns.message.make_query(name, dns.rdatatype.SOA)
     for attempt in range(1, attempts + 1):
         response = dns.query.tcp(
@@ -513,6 +518,7 @@ def ensure_soa_serial_min(
 
 
 def get_zonefile(domain) -> bytes:
+    """Fetch an AXFR and render a filtered zonefile payload."""
     keyring, keyname, keyalgorithm = _transfer_keyring()
     zone_name = domain.name.rstrip(".") + "."
     xfr = dns.query.xfr(
@@ -549,6 +555,7 @@ def get_zonefile(domain) -> bytes:
 
 
 def get_keys(domain):
+    """Return DNSKEYs for a domain, including DS records where applicable."""
     query = dns.message.make_query(domain.name, dns.rdatatype.DNSKEY, want_dnssec=True)
     response = dns.query.tcp(
         query,