feat(api): add perm_create_domain / perm_delete_domain

Related: #885

api/desecapi/migrations/0039_token_perm_create_domain_token_perm_delete_domain.py

@@ -0,0 +1,36 @@
+# Generated by Django 5.1.2 on 2024-11-01 14:29
+
+import django.db.migrations.operations.special
+from django.db import migrations, models
+
+
+def forwards_func(apps, schema_editor):
+    Token = apps.get_model("desecapi", "Token")
+    db_alias = schema_editor.connection.alias
+    Token.objects.using(db_alias).filter(domain_policies__isnull=False).update(
+        perm_create_domain=True, perm_delete_domain=True
+    )
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("desecapi", "0038_user_throttle_daily_rate"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="token",
+            name="perm_create_domain",
+            field=models.BooleanField(default=False),
+        ),
+        migrations.AddField(
+            model_name="token",
+            name="perm_delete_domain",
+            field=models.BooleanField(default=False),
+        ),
+        migrations.RunPython(
+            code=forwards_func,
+            reverse_code=django.db.migrations.operations.special.RunPython.noop,
+        ),
+    ]

api/desecapi/models/tokens.py

@@ -41,6 +41,8 @@ def _allowed_subnets_default():
     name = models.CharField("Name", blank=True, max_length=64)
     last_used = models.DateTimeField(null=True, blank=True)
     mfa = models.BooleanField(default=None, null=True)
+    perm_create_domain = models.BooleanField(default=False)
+    perm_delete_domain = models.BooleanField(default=False)
     perm_manage_tokens = models.BooleanField(default=False)
     allowed_subnets = ArrayField(
         CidrAddressField(), default=_allowed_subnets_default.__func__

api/desecapi/permissions.py

@@ -76,15 +76,6 @@ def has_permission(self, request, view):
         return request.user == view.domain.owner
 
 
-class TokenNoDomainPolicy(permissions.BasePermission):
-    """
-    Permission to check whether a token is unrestricted by any policy.
-    """
-
-    def has_permission(self, request, view):
-        return not request.auth.tokendomainpolicy_set.exists()
-
-
 class TokenHasRRsetPermission(permissions.BasePermission):
     """
     Permission to check whether a token authorizes writing the view's RRset.
@@ -121,6 +112,39 @@ def has_permission(self, request, view):
         return ip in IPv4Network("10.8.0.0/24")
 
 
+class HasCreateDomainPermission(permissions.BasePermission):
+    """
+    Permission to check whether a token has "create domain" permission.
+    """
+
+    def has_permission(self, request, view):
+        return request.auth.perm_create_domain
+
+
+class HasDeleteDomainPermission(permissions.BasePermission):
+    """
+    Permission to check whether a token has "delete domian" permission.
+    """
+
+    def has_permission(self, request, view):
+        return request.auth.perm_delete_domain
+
+    def has_object_permission(self, request, view, obj):
+        policy_set = request.auth.tokendomainpolicy_set
+
+        return (
+            # Ensure token is not explicitly prohibited from writing some RRsets in this domain
+            not policy_set.filter(domain=obj, perm_write=False).exists()
+            and
+            # Ensure applicable domain-independent policies are not constraining
+            (
+                policy_set.filter(domain=obj, subname=None, type=None).exists()
+                # Conservative (might be shadowed by domain=obj for same subname/type)
+                or not policy_set.filter(domain=None, perm_write=False).exists()
+            )
+        )
+
+
 class HasManageTokensPermission(permissions.BasePermission):
     """
     Permission to check whether a token has "manage tokens" permission.

api/desecapi/serializers/tokens.py

@@ -19,6 +19,8 @@ class Meta:
             "max_age",
             "max_unused_period",
             "name",
+            "perm_create_domain",
+            "perm_delete_domain",
             "perm_manage_tokens",
             "allowed_subnets",
             "is_valid",

api/desecapi/tests/base.py

@@ -1352,7 +1352,9 @@ def setUpTestDataWithPdns(cls):
         cls.create_rr_set(cls.my_domain, ["127.0.0.1", "3.2.2.3"], type="A", ttl=123)
         cls.create_rr_set(cls.other_domain, ["40.1.1.1"], type="A", ttl=456)
 
-        cls.token = cls.create_token(user=cls.owner)
+        cls.token = cls.create_token(
+            user=cls.owner, perm_create_domain=True, perm_delete_domain=True
+        )
 
     def setUp(self):
         super().setUp()

api/desecapi/tests/test_domains.py

@@ -1,3 +1,5 @@
+from contextlib import nullcontext
+
 from django.conf import settings
 from django.core import mail
 from django.core.exceptions import ValidationError
@@ -279,6 +281,61 @@ def test_delete_my_domain(self):
         response = self.client.get(url)
         self.assertStatus(response, status.HTTP_404_NOT_FOUND)
 
+    def test_delete_my_domain_policy_constraints(self):
+        policy_sets = [
+            ([dict(domain=None, subname=None, type=None, perm_write=True)], True),
+            ([dict(domain=None, subname=None, type=None, perm_write=False)], False),
+            (
+                [
+                    dict(domain=None, subname=None, type=None, perm_write=False),
+                    dict(
+                        domain=self.my_domain, subname=None, type=None, perm_write=True
+                    ),
+                ],
+                True,
+            ),
+            (
+                [
+                    dict(domain=None, subname=None, type=None, perm_write=False),
+                    dict(
+                        domain=self.my_domain, subname=None, type=None, perm_write=True
+                    ),
+                    dict(
+                        domain=self.my_domain,
+                        subname="_acme-challenge",
+                        type=None,
+                        perm_write=False,
+                    ),
+                ],
+                False,
+            ),
+        ]
+        for policies, permitted in policy_sets:
+            for policy in policies:
+                self.token.tokendomainpolicy_set.create(**policy)
+
+            url = self.reverse("v1:domain-detail", name=self.my_domain.name)
+            with (
+                self.assertRequests(
+                    self.requests_desec_domain_deletion(domain=self.my_domain)
+                )
+                if permitted
+                else nullcontext()
+            ):
+                response = self.client.delete(url)
+                self.assertStatus(
+                    response,
+                    (
+                        status.HTTP_204_NO_CONTENT
+                        if permitted
+                        else status.HTTP_403_FORBIDDEN
+                    ),
+                )
+
+            # Clean-up
+            self.token.tokendomainpolicy_set.all().delete()
+            self.my_domain.save()
+
     def test_delete_other_domain(self):
         url = self.reverse("v1:domain-detail", name=self.other_domain.name)
         response = self.client.delete(url)
@@ -379,6 +436,15 @@ def test_create_domains(self):
             self.assertFalse(domain.is_locally_registrable)
             self.assertEqual(domain.renewal_state, Domain.RenewalState.IMMORTAL)
 
+    def test_create_domain_no_permission(self):
+        self.token.perm_create_domain = False
+        self.token.save()
+
+        response = self.client.post(
+            self.reverse("v1:domain-list"), {"name": "foobar.example"}
+        )
+        self.assertStatus(response, status.HTTP_403_FORBIDDEN)
+
     def test_create_domain_zonefile_import(self):
         zonefile = """$ORIGIN .
 $TTL 43200 ; 12 hours
@@ -829,6 +895,14 @@ def test_delete_my_domain(self):
         response = self.client.get(url)
         self.assertStatus(response, status.HTTP_404_NOT_FOUND)
 
+    def test_delete_my_domain_no_permission(self):
+        self.token.perm_delete_domain = False
+        self.token.save()
+
+        url = self.reverse("v1:domain-detail", name=self.my_domain.name)
+        response = self.client.delete(url)
+        self.assertStatus(response, status.HTTP_403_FORBIDDEN)
+
     def test_delete_other_domains(self):
         url = self.reverse("v1:domain-detail", name=self.other_domain.name)
         with self.assertRequests():

api/desecapi/tests/test_token_domain_policy.py

@@ -458,19 +458,6 @@ def _reset_policies(token):
                     setattr(policy, perm, value)
                     policy.save()
 
-                    # Can't create domain
-                    data = {"name": self.random_domain_name()}
-                    response = self.client.post(
-                        self.reverse("v1:domain-list"), data, **kwargs
-                    )
-                    self.assertStatus(response, status.HTTP_403_FORBIDDEN)
-
-                    # Can't delete domain
-                    response = self.client.delete(
-                        self.reverse("v1:domain-detail", name=domain), {}, **kwargs
-                    )
-                    self.assertStatus(response, status.HTTP_403_FORBIDDEN)
-
                     # Can't access account details
                     response = self.client.get(self.reverse("v1:account"), **kwargs)
                     self.assertStatus(response, status.HTTP_403_FORBIDDEN)
@@ -597,7 +584,7 @@ def test_domain_owner_equals_token_user(self):
             with transaction.atomic():  # https://stackoverflow.com/a/23326971/6867099
                 self.token.save()
 
-    def test_domain_deletion(self):
+    def test_domain_deletion_policy_cleanup(self):
         domains = [None] + self.my_domains[:2]
         for domain in domains:
             models.TokenDomainPolicy(

api/desecapi/tests/test_tokens.py

@@ -59,6 +59,8 @@ def test_retrieve_my_token(self):
                 "max_age",
                 "max_unused_period",
                 "name",
+                "perm_create_domain",
+                "perm_delete_domain",
                 "perm_manage_tokens",
                 "allowed_subnets",
                 "is_valid",
@@ -111,6 +113,8 @@ def test_create_token(self):
         datas = [
             {},
             {"name": "", "perm_manage_tokens": True},
+            {"name": "", "perm_delete_domain": True},
+            {"name": "", "perm_create_domain": True, "perm_delete_domain": True},
             {"name": "foobar"},
             {"allowed_subnets": ["1.2.3.32/28", "bade::affe/128"]},
         ]
@@ -126,6 +130,8 @@ def test_create_token(self):
                     "max_age",
                     "max_unused_period",
                     "name",
+                    "perm_create_domain",
+                    "perm_delete_domain",
                     "perm_manage_tokens",
                     "allowed_subnets",
                     "is_valid",
@@ -137,10 +143,12 @@ def test_create_token(self):
                 response.data["allowed_subnets"],
                 data.get("allowed_subnets", ["0.0.0.0/0", "::/0"]),
             )
-            self.assertEqual(
-                response.data["perm_manage_tokens"],
-                data.get("perm_manage_tokens", False),
-            )
+            for perm in [
+                "perm_create_domain",
+                "perm_delete_domain",
+                "perm_manage_tokens",
+            ]:
+                self.assertEqual(response.data[perm], data.get(perm, False))
             self.assertIsNone(response.data["last_used"])
             self.assertIsNone(Token.objects.get(pk=response.data["id"]).mfa)
 

api/desecapi/views/domains.py

@@ -40,9 +40,10 @@ def permission_classes(self):
             permissions.IsOwner,
         ]
         if self.action == "create":
+            ret.append(permissions.HasCreateDomainPermission)
             ret.append(permissions.WithinDomainLimit)
-        if self.request.method not in SAFE_METHODS:
-            ret.append(permissions.TokenNoDomainPolicy)
+        if self.action == "destroy":
+            ret.append(permissions.HasDeleteDomainPermission)
         return ret
 
     @property

api/desecapi/views/users.py

@@ -101,6 +101,8 @@ def post(self, request, *args, **kwargs):
         user = self.request.user
         token = Token.objects.create(
             user=user,
+            perm_create_domain=True,
+            perm_delete_domain=True,
             perm_manage_tokens=True,
             max_age=timedelta(days=7),
             max_unused_period=timedelta(hours=1),