wip

Author: nils-wisiol

api/api/settings.py

@@ -7,6 +7,7 @@
 from datetime import timedelta
 
 from django.conf.global_settings import PASSWORD_HASHERS as DEFAULT_PASSWORD_HASHERS
+import dns.resolver
 
 BASE_DIR = os.path.dirname(os.path.dirname(__file__))
 
@@ -160,6 +161,13 @@
 ]
 
 DESECSTACK_DOMAIN = os.environ["DESECSTACK_DOMAIN"]
+RESOLVERS = [
+    '9.9.9.9', '2620:fe::fe',  # Quad9
+    '1.1.1.1', '2606:4700:4700::1111',  # Cloudflare 
+    '8.8.8.8', '2001:4860:4860::8888',  # Google
+]
+resolver = dns.resolver.Resolver(configure=False)
+resolver.nameserver = RESOLVERS
 
 # default NS records
 DEFAULT_NS = [name + "." for name in os.environ["DESECSTACK_NS"].strip().split()]

api/desecapi/models/domains.py

@@ -1,8 +1,13 @@
 from __future__ import annotations
 
 from functools import cached_property
+from functools import cache
+from socket import getaddrinfo
 
-import dns
+import dns.name
+import dns.rdataclass
+import dns.rdatatype
+import dns.resolver
 import psl_dns
 from django.conf import settings
 from django.contrib.auth.models import AnonymousUser
@@ -52,6 +57,19 @@ class RenewalState(models.IntegerChoices):
         NOTIFIED = 2
         WARNED = 3
 
+    class DelegationStatus(models.IntegerChoices):
+        NOT_DELEGATED = 0
+        ELSEWHERE = 1
+        PARTIAL = 2
+        EXCLUSIVE = 3
+        MULTI = 4
+
+    class SecurityStatus(models.IntegerChoices):
+        INSECURE = 0
+        BOGUS = 1
+        SECURE_EXCLUSIVE = 2
+        SECURE = 3
+
     created = models.DateTimeField(auto_now_add=True)
     name = models.CharField(
         max_length=191, unique=True, validators=validate_domain_name
@@ -177,6 +195,114 @@ def is_registrable(self):
 
         return True
 
+    @cache
+    def dns_auth_ns(self) -> set[dns.name.Name]:
+        """Queries the DNS to determine the names of authoritative NS 
+        servers for this domain."""
+        name = dns.name.from_text(self.name)
+        
+        # determine parent
+        parent = name.parent()
+        while len(parent):
+            ns_response = settings.resolver.resolve(parent, dns.rdatatype.NS)
+            if ns_response.answer:
+                break
+            parent = parent.parent()
+
+        # extract NS records at the parent
+        try:
+            ns = ns_response.find_rrset(ns_response.answer, parent, dns.rdataclass.IN, dns.rdatatype.NS)
+        except KeyError:  # TODO see if dnspython has an API that includes the except clause already
+            ns = set()
+
+        # look up addresses for parent's NS records
+        addr = {ip for rr in ns for ip in self._lookup(rr.target)}
+
+        # query these NS to determine authoritative NS names
+        self.resolver = dns.resolver.Resolver()
+        self.resolver.nameserver = list(addr)
+        try:
+            ns_authoritative = dns.resolver.resolve(dns.name.from_text(self.name), dns.rdatatype.NS)
+            auth_ns_names = {name.target for name in ns_authoritative}
+        except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN):
+            auth_ns_names = set()
+
+        return auth_ns_names
+
+    @cache  # located at object-level to start with clear cache for new objects
+    def _lookup(target) -> set[str]:
+        try:
+            addrinfo = getaddrinfo(str(target), None)
+        except OSError:
+            return set()
+        return {v[-1][0] for v in addrinfo}    
+
+    @cache
+    def dns_auth_ds(self) -> set:  # TODO update type hint
+        """Queries the DNS to determine the authoritative DS records
+        of this domain."""
+        name = dns.name.from_text(self.name)
+        res = settings.resolver.resolve(name, dns.rdatatype.DS)
+        try:
+            ds = res.find_rrset(
+                res.answer, name, dns.rdataclass.IN, dns.rdatatype.DS
+            )
+        except KeyError:
+            ds = set()
+        return {rr.to_text() for rr in ds}  # TODO use dnspython objects instead of str
+
+    def update_dns_delegation_status(self) -> DelegationStatus:
+        """Queries the DNS to determine the delegation status of this domian and 
+        update the delegation status on record."""
+        auth_ns_names = self.dns_auth_ns()
+        our_ns_names = {dns.name.from_text(ns) for ns in settings.DEFAULT_NS}
+
+        if our_ns_names == auth_ns_names:
+            # just ours
+            self.delegation_status = self.DelegationStatus.EXCLUSIVE
+        elif our_ns_names < auth_ns_names:
+            # all of ours, and others
+            self.delegation_status = self.DelegationStatus.MULTI
+        elif our_ns_names & auth_ns_names:
+            # some of ours, and others
+            self.delegation_status = self.DelegationStatus.PARTIAL
+        elif auth_ns_names:
+            # none of ours, but not empty
+            self.delegation_status = self.DelegationStatus.ELSEWHERE
+        else:
+            # empty
+            self.delegation_status = self.DelegationStatus.NOT_DELEGATED
+
+        return self.delegation_status
+
+    def update_dns_security_status(self) -> SecurityStatus:
+        """Queries the DNS to determine the security status of this domain and 
+        updates the security status on record."""
+        if self.delegation_status not in [self.DelegationStatus.MULTI, self.DelegationStatus.EXCLUSIVE]:
+            self.security_status = None
+            return None
+
+        auth_ds_set = self.dns_auth_ds()
+
+        # Compute overlap of delegation DS records with ours
+        our_ds_set = set()
+        for key in self.keys:  # TODO upgrade this to dnspython objects
+            # Only digest type 2 is mandatory to implement; delegation only fully set up if present
+            our_ds_set |= {
+                ds.lower() for ds in key["ds"] if ds.split(" ")[2] == "2"
+            }
+
+        if our_ds_set == auth_ds_set:
+            self.security_status = self.SecurityStatus.SECURE_EXCLUSIVE
+        elif our_ds_set < auth_ds_set:
+            self.security_status = self.SecurityStatus.SECURE
+        elif auth_ds_set == set():
+            self.security_status = self.SecurityStatus.INSECURE
+        else:
+            self.security_status = self.SecurityStatus.BOGUS
+
+        return self.security_status
+
     @property
     def keys(self):
         if not self._keys: