feat(api): initial version of check-delegation management command

README.md

@@ -334,6 +334,7 @@ While there are certainly many ways to get started hacking desec-stack, here is
            source venv/bin/activate
            pip install wheel
            pip install -r requirements.txt
+           pip install -r requirements-test.txt
 
     1. At this point, Django is ready to run in the virtual environment created above.
         There are two things to consider when running Django outside the container.
@@ -359,10 +360,9 @@ While there are certainly many ways to get started hacking desec-stack, here is
 
            docker compose -f docker-compose.yml -f docker-compose.test-api.yml up -d dbapi
 
-        Finally, you can manage Django using the `manage.py` CLI.
-        As an example, to run the tests, use
+        To run the tests, in `/api` use
 
-           python3 manage.py test
+           pytest
 
     1. Open the project root directory `desec-stack` in PyCharm and select File › Settings.
         1. In Project: desec-stack › Project Structure, mark the `api/` folder as a source folder.

api/Dockerfile

@@ -11,11 +11,12 @@ WORKDIR /usr/src/app
 ENV PIP_DISABLE_PIP_VERSION_CHECK=1
 ENV PIP_NO_CACHE_DIR=1
 
-COPY requirements.txt /usr/src/app/
+COPY requirements*.txt /usr/src/app/
 # freetype-dev is needed for captcha generation
 RUN apk add --no-cache gcc freetype-dev libffi-dev musl-dev libmemcached-dev postgresql-dev jpeg-dev zlib-dev
 RUN pip install --upgrade pip \
     && pip install -r requirements.txt \
+    && pip install -r requirements-test.txt \
     && pip freeze
 
 RUN mkdir /root/cronhook

api/api/settings.py

@@ -160,6 +160,14 @@
 ]
 
 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
+]
 
 # default NS records
 DEFAULT_NS = [name + "." for name in os.environ["DESECSTACK_NS"].strip().split()]

api/desecapi/migrations/0045_domain_delegation_status_and_more.py

@@ -0,0 +1,72 @@
+# Generated by Django 5.1.8 on 2025-04-12 15:02
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("desecapi", "0044_alter_captcha_created_alter_domain_renewal_state_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="domain",
+            name="delegation_status",
+            field=models.IntegerField(
+                blank=True,
+                choices=[
+                    (0, "Not Delegated"),
+                    (1, "Elsewhere"),
+                    (2, "Partial"),
+                    (3, "Exclusive"),
+                    (4, "Multi"),
+                    (128, "Error Nxdomain"),
+                    (129, "Error No Answer"),
+                    (130, "Error No Nameservers"),
+                    (131, "Error Timeout"),
+                ],
+                default=None,
+                null=True,
+            ),
+        ),
+        migrations.AddField(
+            model_name="domain",
+            name="delegation_status_changed",
+            field=models.DateTimeField(blank=True, default=None, null=True),
+        ),
+        migrations.AddField(
+            model_name="domain",
+            name="delegation_status_touched",
+            field=models.DateTimeField(blank=True, default=None, null=True),
+        ),
+        migrations.AddField(
+            model_name="domain",
+            name="security_status",
+            field=models.IntegerField(
+                blank=True,
+                choices=[
+                    (0, "Insecure"),
+                    (1, "Foreign Keys"),
+                    (2, "Secure Exclusive"),
+                    (3, "Secure"),
+                    (128, "Error Nxdomain"),
+                    (129, "Error No Answer"),
+                    (130, "Error No Nameservers"),
+                    (131, "Error Timeout"),
+                ],
+                default=None,
+                null=True,
+            ),
+        ),
+        migrations.AddField(
+            model_name="domain",
+            name="security_status_changed",
+            field=models.DateTimeField(blank=True, default=None, null=True),
+        ),
+        migrations.AddField(
+            model_name="domain",
+            name="security_status_touched",
+            field=models.DateTimeField(blank=True, default=None, null=True),
+        ),
+    ]

api/desecapi/models/domains.py

@@ -1,15 +1,21 @@
 from __future__ import annotations
 
-from functools import cached_property
-
-import dns
+from functools import cache, cached_property
+from socket import getaddrinfo
+
+import dns.name
+import dns.rdataclass
+import dns.rdatatype
+import dns.rdtypes
+import dns.resolver
 import psl_dns
 from django.conf import settings
 from django.contrib.auth.models import AnonymousUser
 from django.core.exceptions import ValidationError
 from django.db import models
 from django.db.models import CharField, F, Manager, Q, Value
 from django.db.models.functions import Concat, Length
+from django.utils import timezone
 from django_prometheus.models import ExportModelOperationsMixin
 from dns.exception import Timeout
 from dns.resolver import NoNameservers
@@ -23,6 +29,13 @@
 
 psl = psl_dns.PSL(resolver=settings.PSL_RESOLVER, timeout=0.5)
 
+# CHECKING DISABLED general-purpose resolver for queries to the public DNS
+resolver_CD = dns.resolver.Resolver(configure=False)
+resolver_CD.nameservers = settings.RESOLVERS
+resolver_CD.flags = (
+    (resolver_CD.flags or 0) | dns.flags.CD | dns.flags.AD | dns.flags.RD
+)
+
 
 class DomainManager(Manager):
     def filter_qname(self, qname: str, **kwargs) -> models.query.QuerySet:
@@ -52,6 +65,25 @@ class RenewalState(models.IntegerChoices):
         NOTIFIED = 2
         WARNED = 3
 
+    class DelegationStatus(models.IntegerChoices):
+        NOT_DELEGATED = 0
+        ELSEWHERE = 1
+        PARTIAL = 2
+        EXCLUSIVE = 3
+        MULTI = 4
+        ERROR_NXDOMAIN = 128
+        ERROR_NO_NAMESERVERS = 129
+        ERROR_TIMEOUT = 130
+
+    class SecurityStatus(models.IntegerChoices):
+        INSECURE = 0
+        FOREIGN_KEYS = 1
+        SECURE_EXCLUSIVE = 2
+        SECURE = 3
+        ERROR_NXDOMAIN = 128
+        ERROR_NO_NAMESERVERS = 130
+        ERROR_TIMEOUT = 131
+
     created = models.DateTimeField(auto_now_add=True)
     name = models.CharField(
         max_length=191, unique=True, validators=validate_domain_name
@@ -63,6 +95,26 @@ class RenewalState(models.IntegerChoices):
         choices=RenewalState.choices, db_index=True, default=RenewalState.IMMORTAL
     )
     renewal_changed = models.DateTimeField(auto_now_add=True)
+    delegation_status = models.IntegerField(
+        choices=DelegationStatus.choices,
+        default=None,
+        null=True,
+        blank=True,
+    )
+    delegation_status_touched = models.DateTimeField(
+        default=None, null=True, blank=True
+    )
+    delegation_status_changed = models.DateTimeField(
+        default=None, null=True, blank=True
+    )
+    security_status = models.IntegerField(
+        choices=SecurityStatus.choices,
+        default=None,
+        null=True,
+        blank=True,
+    )
+    security_status_touched = models.DateTimeField(default=None, null=True, blank=True)
+    security_status_changed = models.DateTimeField(default=None, null=True, blank=True)
 
     _keys = None
     objects = DomainManager()
@@ -177,6 +229,176 @@ def is_registrable(self):
 
         return True
 
+    @staticmethod
+    @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}
+
+    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.
+
+        The delegation status is evaluated by trying to determine the NS records
+        as defined in the parent zone to the this `Domain`.
+
+        The parent zone's apex is determined by walking up the DNS tree, querying
+        NS records until an answer is provided. Once the apex and nameservers are
+        found, they are queried for the nameservers of `self.name`.
+        """
+        old_delegation_status = self.delegation_status
+
+        our_ns_addrs = {
+            rr.address
+            for name in settings.DEFAULT_NS
+            for rrtype in {"A", "AAAA"}
+            for rr in list(resolver_CD.resolve(name, rrtype))
+        }
+
+        try:
+            # Determine the parent zone nameservers
+            ## names
+            parent_apex_candidate = dns.name.from_text(self.name).parent()
+            parent_nameservers = []
+            while len(parent_apex_candidate) > 1:
+                parent_nameservers = resolver_CD.resolve(
+                    parent_apex_candidate, dns.rdatatype.NS, raise_on_no_answer=False
+                )
+                if parent_nameservers:
+                    break
+                else:
+                    parent_apex_candidate = parent_apex_candidate.parent()
+
+            # TODO what if no parnet_nameservers? what if len(parent_apex_candidate) == 0?
+
+            ## addresses
+            parent_nameservers_addrs = [
+                rr.address
+                for ns in parent_nameservers
+                for rrtype in {"A", "AAAA"}
+                for rr in list(
+                    resolver_CD.resolve(ns.target, rrtype, raise_on_no_answer=False)
+                )
+            ]  # TODO duplicate with above
+
+            # Determine this zone's nameservers as defined at parent zone nameservers
+            ## names
+            resolver = dns.resolver.Resolver(configure=False)
+            resolver.nameservers = parent_nameservers_addrs
+            resolver.flags = dns.flags.AD
+            auth_ns_names = {
+                rr.target
+                for rr in resolver.resolve(
+                    self.name, dns.rdatatype.NS, raise_on_no_answer=False
+                )
+            }  # FIXME does not work because NS records are located in the AUTHORITY section instead of ANSWER section
+
+            ## addresses
+            auth_ns_addrs = {
+                rr.address
+                for name in auth_ns_names
+                for rrtype in {"A", "AAAA"}
+                for rr in list(
+                    resolver_CD.resolve(name, rrtype, raise_on_no_answer=False)
+                )
+            }  # TODO duplicate with above
+        except dns.resolver.NXDOMAIN:
+            self.delegation_status = self.DelegationStatus.ERROR_NXDOMAIN
+        except dns.resolver.NoNameservers:
+            self.delegation_status = self.DelegationStatus.ERROR_NO_NAMESERVERS
+        except dns.resolver.LifetimeTimeout:
+            self.delegation_status = self.DelegationStatus.ERROR_TIMEOUT
+        else:
+
+            if our_ns_addrs == auth_ns_addrs:
+                # just ours
+                self.delegation_status = self.DelegationStatus.EXCLUSIVE
+            elif our_ns_addrs < auth_ns_addrs:
+                # all of ours, and others
+                self.delegation_status = self.DelegationStatus.MULTI
+            elif our_ns_addrs & auth_ns_addrs:
+                # intersection is non-empty, but not all of our's are included
+                # some of ours, and others
+                self.delegation_status = self.DelegationStatus.PARTIAL
+            elif auth_ns_addrs:
+                # none of ours, but not empty
+                self.delegation_status = self.DelegationStatus.ELSEWHERE
+            elif auth_ns_addrs == set():
+                # empty
+                self.delegation_status = self.DelegationStatus.NOT_DELEGATED
+            elif auth_ns_addrs is None:
+                # error
+                self.delegation_status = self.DelegationStatus
+
+            print(
+                self.name,
+                list(parent_nameservers),
+                list(parent_nameservers_addrs),
+                auth_ns_names,
+                auth_ns_addrs,
+            )
+
+        now = timezone.now()
+        self.delegation_status_touched = now
+        if old_delegation_status != self.delegation_status:
+            self.delegation_status_changed = now
+        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."""
+        old_security_status = self.security_status
+
+        if self.delegation_status not in [
+            self.DelegationStatus.MULTI,
+            self.DelegationStatus.EXCLUSIVE,
+        ]:
+            self.security_status = None
+            return None
+
+        try:
+            auth_ds = set(
+                resolver_CD.resolve(
+                    self.name, dns.rdatatype.DS, raise_on_no_answer=False
+                )
+            )
+        except dns.resolver.NXDOMAIN:
+            self.security_status = self.SecurityStatus.ERROR_NXDOMAIN
+        except dns.resolver.NoNameservers:
+            self.delegation_status = self.SecurityStatus.ERROR_NO_NAMESERVERS
+        except dns.resolver.LifetimeTimeout:
+            self.delegation_status = self.SecurityStatus.ERROR_TIMEOUT
+        else:
+            auth_ds = {ds for ds in auth_ds if ds.digest_type == 2}
+
+            # Compute overlap of delegation DS records with ours
+            our_ds_set = {
+                dns.rdata.from_text(rdclass="IN", rdtype="DS", tok=ds)
+                for key in self.keys
+                for ds in key.get("ds", [])
+                if dns.rdata.from_text(rdclass="IN", rdtype="DS", tok=ds).digest_type
+                == 2  # only digest type 2 is mandatory
+            }
+
+            if our_ds_set == auth_ds:
+                self.security_status = self.SecurityStatus.SECURE_EXCLUSIVE
+            elif our_ds_set < auth_ds:
+                self.security_status = self.SecurityStatus.SECURE
+            elif auth_ds != set():
+                self.security_status = self.SecurityStatus.FOREIGN_KEYS
+            else:
+                self.security_status = self.SecurityStatus.INSECURE
+
+        now = timezone.now()
+        self.security_status_touched = now
+        if old_security_status != self.security_status:
+            self.security_status_changed = now
+        return self.security_status
+
     @property
     def keys(self):
         if not self._keys: