sven/netbox-plugin-dns-bridge
0
0 Fork 0
Code Actions Packages Releases Wiki Activity

Refactor: move shared NOTIFY helpers into a NotifyJobRunner base class

Browse Source 
- New NotifyJobRunner base class (extends netbox.jobs.JobRunner) holds
  the plugin SETTINGS and the helpers shared by both NOTIFY jobs:
  load_tsig_key(), send_notify() and get_soa_serial() (promoted from
  the former module-level _-prefixed functions).
- SendClientNotify and SendNSNotify extend the base and access the
  helpers via self. Behavior, job names and kwargs are unchanged.
- The nameserver-only helpers moved into SendNSNotify as
  notify_ns_enabled() (classmethod, reads cls.SETTINGS) and
  resolve_zone_nameserver_ips() (instance method).
- No changes to any callers: schedule_client_notify(),
  schedule_ns_notify() and the catalog zone NOTIFY scheduling keep
  their names and call sites.
This commit is contained in:
sven committed 2026-09-22 23:15:09 +02:00
1 parent fe24dbdd98
commit c15bdfca05
1 file changed
+125 -122
Download Patch File Download Diff File Expand all files Collapse all files
netbox_dns_bridge/jobs/notify.py
+125 -122
View File
@@ -15,11 +15,15 @@ from netbox_dns_bridge.utils import get_logger
from netbox_dns_bridge.models import SeenTransferClient
LOGGER = get_logger(__name__)
SETTINGS = settings.PLUGINS_CONFIG.get("netbox_dns_bridge", {})
def _load_tsig_key(view_name: str) -> dns.tsig.Key:
tsig_keys = SETTINGS.get("tsig_keys", {})
class NotifyJobRunner(JobRunner):
"""Base class for NOTIFY jobs: shared plugin settings and helpers."""
SETTINGS = settings.PLUGINS_CONFIG.get("netbox_dns_bridge", {})
def load_tsig_key(self, view_name: str) -> dns.tsig.Key:
tsig_keys = self.SETTINGS.get("tsig_keys", {})
if view_name not in tsig_keys:
raise RuntimeError(f"No TSIG key configured for view '{view_name}'")
@@ -41,8 +45,8 @@ def _load_tsig_key(view_name: str) -> dns.tsig.Key:
return dns.tsig.Key(name=key_name, secret=secret, algorithm=algorithm)
def _send_notify(
def send_notify(
self,
zone_name,
soa_serial,
targets,
@@ -55,7 +59,7 @@ def _send_notify(
soa_retry=10,
soa_expire=1209600,
soa_minimum=0,
):
):
"""Build a DNS NOTIFY message for *zone_name* and send it to every target IP."""
fqdn = dns.name.from_text(zone_name, dns.name.root).to_text()
msg = dns.message.make_query(fqdn, dns.rdatatype.SOA)
@@ -88,99 +92,8 @@ def _send_notify(
f" {view_name}/{fqdn} {soa_serial}: {e}"
)
class SendClientNotify(JobRunner):
class Meta:
name = "Send DNS NOTIFY"
def run(self, *args, **kwargs):
notify_over_tcp = SETTINGS.get("notify_over_tcp", False)
port = SETTINGS.get("notify_client_port", 53)
view_name = kwargs["view_name"]
zone_name = kwargs["zone_name"]
soa_serial = kwargs["soa_serial"]
soa_refresh = kwargs.get("soa_refresh", 60)
soa_retry = kwargs.get("soa_retry", 10)
soa_expire = kwargs.get("soa_expire", 1209600)
soa_minimum = kwargs.get("soa_minimum", 0)
try:
view = View.objects.get(name=view_name)
tsig_key = _load_tsig_key(view_name)
cutoff_hours = SETTINGS.get("notify_client_alive_threshold_hours", 24)
seen_cutoff = timezone.now() - timezone.timedelta(hours=cutoff_hours)
clients = list(
SeenTransferClient.objects.filter(
view=view, last_seen__gte=seen_cutoff
).values_list("source_ip", flat=True)
)
_send_notify(
zone_name, soa_serial, clients, tsig_key,
view_name, notify_over_tcp, port, log_prefix="NOTIFY",
soa_refresh=soa_refresh, soa_retry=soa_retry,
soa_expire=soa_expire, soa_minimum=soa_minimum,
)
except View.DoesNotExist as e:
LOGGER.error(f"Failed to get view for NOTIFY: {e}")
except OperationalError as e:
LOGGER.error(f"NOTIFY failed due to unexpected error: {e}")
close_old_connections()
class SendNSNotify(JobRunner):
"""Send a NOTIFY straight to a zone's own nameservers.
Unlike ``SendClientNotify`` (which notifies clients that previously queried
the transfer endpoint), this job resolves the addresses of the zone's
configured nameservers from NetBox DNS itself and notifies each of them so
they pick up the changed zone quickly.
"""
class Meta:
name = "Send DNS NOTIFY to zone nameservers"
def run(self, *args, **kwargs):
notify_over_tcp = SETTINGS.get("notify_over_tcp", False)
port = SETTINGS.get("notify_ns_port", 53)
view_name = kwargs["view_name"]
zone_name = kwargs["zone_name"]
soa_serial = kwargs["soa_serial"]
try:
zone = Zone.objects.get(pk=kwargs["zone_id"])
except Zone.DoesNotExist:
LOGGER.error(f"NS NOTIFY: zone id {kwargs['zone_id']} no longer exists")
return
except OperationalError as e:
LOGGER.error(f"NS NOTIFY failed due to unexpected error: {e}")
close_old_connections()
return
ns_targets = _resolve_zone_nameserver_ips(zone)
if not ns_targets:
LOGGER.warning(
f"NS NOTIFY: no nameserver addresses found in NetBox DNS for "
f"{view_name}/{zone_name} — skipping"
)
return
try:
tsig_key = _load_tsig_key(view_name)
except RuntimeError as e:
LOGGER.error(f"NS NOTIFY aborted for {view_name}/{zone_name}: {e}")
return
_send_notify(
zone_name, soa_serial, ns_targets, tsig_key,
view_name, notify_over_tcp, port, log_prefix="NS NOTIFY",
)
def _get_soa_serial(zone: Zone):
@staticmethod
def get_soa_serial(zone: Zone):
"""Return the current SOA serial of a zone parsed from its SOA record."""
try:
soa_record = Record.objects.filter(zone=zone, type="SOA", active=True).first()
@@ -199,7 +112,115 @@ def _get_soa_serial(zone: Zone):
return None
def _resolve_zone_nameserver_ips(zone: Zone) -> list:
class SendClientNotify(NotifyJobRunner):
class Meta:
name = "Send DNS NOTIFY"
def run(self, *args, **kwargs):
notify_over_tcp = self.SETTINGS.get("notify_over_tcp", False)
port = self.SETTINGS.get("notify_client_port", 53)
view_name = kwargs["view_name"]
zone_name = kwargs["zone_name"]
soa_serial = kwargs["soa_serial"]
soa_refresh = kwargs.get("soa_refresh", 60)
soa_retry = kwargs.get("soa_retry", 10)
soa_expire = kwargs.get("soa_expire", 1209600)
soa_minimum = kwargs.get("soa_minimum", 0)
try:
view = View.objects.get(name=view_name)
tsig_key = self.load_tsig_key(view_name)
cutoff_hours = self.SETTINGS.get("notify_client_alive_threshold_hours", 24)
seen_cutoff = timezone.now() - timezone.timedelta(hours=cutoff_hours)
clients = list(
SeenTransferClient.objects.filter(
view=view, last_seen__gte=seen_cutoff
).values_list("source_ip", flat=True)
)
self.send_notify(
zone_name, soa_serial, clients, tsig_key,
view_name, notify_over_tcp, port, log_prefix="NOTIFY",
soa_refresh=soa_refresh, soa_retry=soa_retry,
soa_expire=soa_expire, soa_minimum=soa_minimum,
)
except View.DoesNotExist as e:
LOGGER.error(f"Failed to get view for NOTIFY: {e}")
except OperationalError as e:
LOGGER.error(f"NOTIFY failed due to unexpected error: {e}")
close_old_connections()
class SendNSNotify(NotifyJobRunner):
"""Send a NOTIFY straight to a zone's own nameservers.
Unlike ``SendClientNotify`` (which notifies clients that previously queried
the transfer endpoint), this job resolves the addresses of the zone's
configured nameservers from NetBox DNS itself and notifies each of them so
they pick up the changed zone quickly.
"""
class Meta:
name = "Send DNS NOTIFY to zone nameservers"
def run(self, *args, **kwargs):
notify_over_tcp = self.SETTINGS.get("notify_over_tcp", False)
port = self.SETTINGS.get("notify_ns_port", 53)
view_name = kwargs["view_name"]
zone_name = kwargs["zone_name"]
soa_serial = kwargs["soa_serial"]
try:
zone = Zone.objects.get(pk=kwargs["zone_id"])
except Zone.DoesNotExist:
LOGGER.error(f"NS NOTIFY: zone id {kwargs['zone_id']} no longer exists")
return
except OperationalError as e:
LOGGER.error(f"NS NOTIFY failed due to unexpected error: {e}")
close_old_connections()
return
ns_targets = self.resolve_zone_nameserver_ips(zone)
if not ns_targets:
LOGGER.warning(
f"NS NOTIFY: no nameserver addresses found in NetBox DNS for "
f"{view_name}/{zone_name} — skipping"
)
return
try:
tsig_key = self.load_tsig_key(view_name)
except RuntimeError as e:
LOGGER.error(f"NS NOTIFY aborted for {view_name}/{zone_name}: {e}")
return
self.send_notify(
zone_name, soa_serial, ns_targets, tsig_key,
view_name, notify_over_tcp, port, log_prefix="NS NOTIFY",
)
@classmethod
def notify_ns_enabled(cls, zone) -> bool:
"""Return True if a NOTIFY to the zone's nameservers should be sent.
Enabled either globally for every zone via ``notify_ns_all_zones`` or per
zone when the boolean custom field named by ``notify_ns_custom_field_name``
is set to True on the zone.
"""
if cls.SETTINGS.get("notify_ns_all_zones", False):
return True
cf_name = cls.SETTINGS.get("notify_ns_custom_field_name")
if not cf_name:
return False
custom_fields = getattr(zone, "custom_field_data", None) or {}
return custom_fields.get(cf_name) is True
def resolve_zone_nameserver_ips(self, zone: Zone) -> list:
"""Resolve the zone's nameserver hostnames to IPs using NetBox DNS records.
For each nameserver assigned to the zone, matching active A/AAAA records
@@ -234,29 +255,11 @@ def _resolve_zone_nameserver_ips(zone: Zone) -> list:
return ips
def _notify_ns_enabled(zone) -> bool:
"""Return True if a NOTIFY to the zone's nameservers should be sent.
Enabled either globally for every zone via ``notify_ns_all_zones`` or per
zone when the boolean custom field named by ``notify_ns_custom_field_name``
is set to True on the zone.
"""
if SETTINGS.get("notify_ns_all_zones", False):
return True
cf_name = SETTINGS.get("notify_ns_custom_field_name")
if not cf_name:
return False
custom_fields = getattr(zone, "custom_field_data", None) or {}
return custom_fields.get(cf_name) is True
def schedule_client_notify(zone: Zone):
if not SETTINGS.get("notify_clients", False):
if not SendClientNotify.SETTINGS.get("notify_clients", False):
return
soa_serial = _get_soa_serial(zone)
soa_serial = SendClientNotify.get_soa_serial(zone)
if soa_serial is None:
return
@@ -268,10 +271,10 @@ def schedule_client_notify(zone: Zone):
def schedule_ns_notify(zone: Zone):
if not _notify_ns_enabled(zone):
if not SendNSNotify.notify_ns_enabled(zone):
return
soa_serial = _get_soa_serial(zone)
soa_serial = SendNSNotify.get_soa_serial(zone)
if soa_serial is None:
return