feat(api): gate NS changes on local suffix

api/desecapi/migrations/0046_domain_allow_local_ns_changes.py

@@ -0,0 +1,37 @@
+# Generated by Django 5.2.10 on 2026-02-10 00:00
+
+from django.conf import settings
+from django.db import migrations, models
+
+
+def set_allow_local_ns_changes(apps, schema_editor):
+    Domain = apps.get_model("desecapi", "Domain")
+    Domain.objects.update(allow_local_ns_changes=True)
+
+    local_suffixes = set(settings.LOCAL_PUBLIC_SUFFIXES)
+    if not local_suffixes:
+        return
+
+    to_update = []
+    for domain in Domain.objects.only("id", "name").iterator():
+        parent = domain.name.partition(".")[2] or None
+        if parent in local_suffixes:
+            to_update.append(domain.id)
+
+    if to_update:
+        Domain.objects.filter(id__in=to_update).update(allow_local_ns_changes=False)
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("desecapi", "0045_rr_unique_record_in_rrset"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="domain",
+            name="allow_local_ns_changes",
+            field=models.BooleanField(default=True),
+        ),
+        migrations.RunPython(set_allow_local_ns_changes, migrations.RunPython.noop),
+    ]

api/desecapi/models/domains.py

@@ -63,6 +63,7 @@ class RenewalState(models.IntegerChoices):
         choices=RenewalState.choices, db_index=True, default=RenewalState.IMMORTAL
     )
     renewal_changed = models.DateTimeField(auto_now_add=True)
+    allow_local_ns_changes = models.BooleanField(default=True)
 
     _keys = None
     objects = DomainManager()
@@ -74,6 +75,7 @@ class Meta:
         ordering = ("created",)
 
     def __init__(self, *args, **kwargs):
+        allow_local_ns_changes_provided = "allow_local_ns_changes" in kwargs
         if isinstance(kwargs.get("owner"), AnonymousUser):
             kwargs = {**kwargs, "owner": None}  # make a copy and override
         # Avoid super().__init__(owner=None, ...) to not mess up *values instantiation in django.db.models.Model.from_db
@@ -85,6 +87,12 @@ def __init__(self, *args, **kwargs):
             and self.is_locally_registrable
         ):
             self.renewal_state = Domain.RenewalState.FRESH
+        if (
+            self.pk is None
+            and not allow_local_ns_changes_provided
+            and self.is_locally_registrable
+        ):
+            self.allow_local_ns_changes = False
 
     @cached_property
     def public_suffix(self):
@@ -226,6 +234,10 @@ def touched(self):
     def is_locally_registrable(self):
         return self.parent_domain_name in settings.LOCAL_PUBLIC_SUFFIXES
 
+    @property
+    def can_modify_ns_records(self):
+        return (not self.is_locally_registrable) or self.allow_local_ns_changes
+
     @property
     def _owner_or_none(self):
         try:

api/desecapi/serializers/records.py

@@ -541,7 +541,7 @@ def validate(self, attrs):
         # Deletion using records=[] is allowed, except at the apex
         if (
             type_ == "NS"
-            and self.domain.is_locally_registrable
+            and not self.domain.can_modify_ns_records
             and (
                 attrs.get("records", True)
                 or not attrs.get("subname", self.instance.subname)

api/desecapi/tests/test_rrsets.py

@@ -1824,3 +1824,88 @@ def test_bulk_delete_ns_rrset_nonapex(self):
                 response = method(self.my_domain.name, [data])
                 self.assertStatus(response, status.HTTP_200_OK)
                 self.assertEqual(response.data, [])
+
+
+class AuthenticatedRRSetLPSNSPermissionTestCase(AuthenticatedRRSetBaseTestCase):
+    DYN = True
+
+    ns_data = {"type": "NS", "records": ["ns.example."], "ttl": 3600}
+
+    def setUp(self):
+        super().setUp()
+        for domain in (self.my_domain, self.my_empty_domain):
+            domain.allow_local_ns_changes = True
+            domain.save(update_fields=["allow_local_ns_changes"])
+
+    def test_create_ns_rrset_allowed(self):
+        for subname in ["", "sub"]:
+            data = dict(self.ns_data, subname=subname)
+            with self.assertRequests(
+                self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
+            ):
+                response = self.client.post_rr_set(
+                    domain_name=self.my_empty_domain.name, **data
+                )
+                self.assertStatus(response, status.HTTP_201_CREATED)
+
+    def test_update_ns_rrset_allowed(self):
+        for subname in ["", "sub"]:
+            for method in (self.client.patch_rr_set, self.client.put_rr_set):
+                create_records = settings.DEFAULT_NS
+                update_records = list(self.ns_data["records"])
+                rrset = self.my_domain.rrset_set.filter(
+                    subname=subname, type="NS"
+                ).first()
+                if rrset is None:
+                    try:
+                        rrset = self.create_rr_set(
+                            self.my_domain,
+                            create_records,
+                            subname=subname,
+                            type="NS",
+                            ttl=3600,
+                        )
+                    except IntegrityError:
+                        rrset = self.my_domain.rrset_set.get(subname=subname, type="NS")
+                rrset.save_records(create_records)
+                current_records = list(rrset.records.values_list("content", flat=True))
+                if set(current_records) == set(update_records):
+                    update_records = create_records
+                update_data = dict(
+                    self.ns_data, subname=subname, records=update_records
+                )
+                with self.assertRequests(
+                    self.requests_desec_rr_sets_update(name=self.my_domain.name)
+                ):
+                    response = method(self.my_domain.name, subname, "NS", update_data)
+                    self.assertStatus(response, status.HTTP_200_OK)
+
+    def test_delete_ns_rrset_apex_allowed(self):
+        data = dict(self.ns_data, records=[], subname="")
+        for method in (self.client.patch_rr_set, self.client.put_rr_set):
+            if not self.my_domain.rrset_set.filter(subname="", type="NS").exists():
+                self.create_rr_set(
+                    self.my_domain,
+                    settings.DEFAULT_NS,
+                    subname="",
+                    type="NS",
+                    ttl=3600,
+                )
+            with self.assertRequests(
+                self.requests_desec_rr_sets_update(name=self.my_domain.name)
+            ):
+                response = method(self.my_domain.name, "", "NS", data)
+                self.assertStatus(response, status.HTTP_204_NO_CONTENT)
+        if not self.my_domain.rrset_set.filter(subname="", type="NS").exists():
+            self.create_rr_set(
+                self.my_domain,
+                settings.DEFAULT_NS,
+                subname="",
+                type="NS",
+                ttl=3600,
+            )
+        with self.assertRequests(
+            self.requests_desec_rr_sets_update(name=self.my_domain.name)
+        ):
+            response = self.client.delete_rr_set(self.my_domain.name, "", "NS")
+            self.assertStatus(response, status.HTTP_204_NO_CONTENT)

api/desecapi/views/records.py

@@ -100,7 +100,7 @@ def update(self, request, *args, **kwargs):
 
     def perform_destroy(self, instance):
         # Disallow modification of apex NS RRset for locally registrable domains
-        if instance.type == "NS" and self.domain.is_locally_registrable:
+        if instance.type == "NS" and not self.domain.can_modify_ns_records:
             if instance.subname == "":
                 raise ValidationError("Cannot modify NS records for this domain.")
         with PDNSChangeTracker():