Feature: Delegate zone to another name server

netbox_ddns/background_tasks.py

@@ -7,10 +7,10 @@
 from django.db import IntegrityError
 from django_rq import job
 from dns import rcode
-from netaddr import ip
+from netaddr import ip, IPNetwork
 
 from netbox_ddns.models import ACTION_CREATE, ACTION_DELETE, DNSStatus, RCODE_NO_ZONE, ReverseZone, Zone
-from netbox_ddns.utils import get_soa
+from netbox_ddns.utils import get_soa, check_servers_authoritative, get_ip
 
 logger = logging.getLogger('netbox_ddns')
 
@@ -198,3 +198,36 @@ def dns_delete(dns_name: str, address: ip.IPAddress, forward=True, reverse=True,
             status.save(force_update=True)
 
     return ', '.join(output)
+
+
+# TODO: dns_create,dns_remove check if it is not delegated to another server
+def create_reverse_delegation(nameservers: List[str], prefix: IPNetwork, status: Optional[DNSStatus], output: List[str]):
+    if prefix.size > 24:
+        output.append(f'Classless delegation is not implemented, prefix size too big: {prefix.cidr}')
+
+    if status:
+        status.reverse_action = ACTION_CREATE
+
+    nameservers = {x: get_ip(x) for x in nameservers}
+
+    subnets = prefix.subnet(24)
+    while 1:
+        subnet = next(subnets, None)
+        if subnet is None:
+            break
+
+        zone = ReverseZone.objects.find_for_address(subnet.network)
+        if zone:
+            for nameserver, addresses in nameservers.items():
+                # TODO: check authority
+                nameserver = {nameserver: address}
+                if check_servers_authoritative(zone.name, addresses):
+                    pass
+            update = zone.server.create_update(zone.name)
+            for nameserver in nameservers.keys():
+                update.add(
+                    zone.name,
+                    zone.ttl,
+                    dns.rdatatype.NS,
+                    nameserver
+                )

netbox_ddns/utils.py

@@ -1,3 +1,5 @@
+from typing import List, Dict
+
 import dns.rdatatype
 import dns.resolver
 
@@ -26,3 +28,33 @@ def get_soa(dns_name: str) -> str:
                 for rrset in response.authority:
                     if rrset.rdtype == dns.rdatatype.SOA:
                         return rrset.name.to_text()
+
+
+def check_servers_authoritative(zone: str, nameservers: Dict[str, List[str]]) -> Dict[str, bool]:
+    resolver = dns.resolver.Resolver()
+    res = {}
+
+    for ns, addr in nameservers.items():
+        try:
+            resolver.nameservers = [addr]
+            answer = resolver.resolve(zone, dns.rdatatype.NS)
+            if ns in answer.rrset.to_text():
+                res[ns] = True
+            else:
+                res[ns] = False
+        except dns.resolver.NoNameservers:
+            res[ns] = False
+
+    return res
+
+
+def get_ip(address: str, family=0) -> List[str]:
+    resolver = dns.resolver.Resolver()
+
+    if family == 0:
+        answer_a = [res.to_text() for res in resolver.resolve(address, dns.rdatatype.A)]
+        answer_aaaa = [res.to_text() for res in resolver.resolve(address, dns.rdatatype.AAAA)]
+        return answer_a + answer_aaaa
+    else:
+        answer = resolver.resolve(address, dns.rdatatype.AAAA if family == 6 else dns.rdatatype.A)
+        return [res.to_text() for res in answer]