feat(api): add Token.auto_policy (including default policy logic)

Related: #885

api/desecapi/migrations/0040_token_auto_policy_token_token_auto_policy_and_more.py

@@ -0,0 +1,50 @@
+# Generated by Django 5.1.3 on 2024-11-21 15:45
+
+import pgtrigger.compiler
+import pgtrigger.migrations
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("desecapi", "0039_token_perm_create_domain_token_perm_delete_domain"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="token",
+            name="auto_policy",
+            field=models.BooleanField(default=False),
+        ),
+        pgtrigger.migrations.AddTrigger(
+            model_name="token",
+            trigger=pgtrigger.compiler.Trigger(
+                name="token_auto_policy",
+                sql=pgtrigger.compiler.UpsertTriggerSql(
+                    constraint="CONSTRAINT",
+                    func="\n                    IF\n                        NEW.auto_policy = true AND NOT EXISTS(\n                            SELECT * FROM desecapi_tokendomainpolicy WHERE token_id = NEW.id AND domain_id IS NULL AND subname IS NULL AND type IS NULL\n                        )\n                    THEN\n                        RAISE EXCEPTION 'Token auto policy without a default policy is not allowed. (token.id=%s)', NEW.id;\n                    END IF;\n                    RETURN NULL;\n                ",
+                    hash="f2cc6e70892817e04cbfbbff63bd4ddbe05b9aa5",
+                    operation="UPDATE OR INSERT",
+                    pgid="pgtrigger_token_auto_policy_8e6d9",
+                    table="desecapi_token",
+                    timing="DEFERRABLE INITIALLY DEFERRED",
+                    when="AFTER",
+                ),
+            ),
+        ),
+        pgtrigger.migrations.AddTrigger(
+            model_name="tokendomainpolicy",
+            trigger=pgtrigger.compiler.Trigger(
+                name="default_policy_when_auto_policy",
+                sql=pgtrigger.compiler.UpsertTriggerSql(
+                    func="\n                    IF\n                        OLD.domain_id IS NULL AND OLD.subname IS NULL AND OLD.type IS NULL AND (SELECT auto_policy FROM desecapi_token WHERE id = OLD.token_id) = true\n                    THEN\n                        RAISE EXCEPTION 'Cannot delete default policy while auto_policy is in effect. (tokendomainpolicy.id=%s)', OLD.id;\n                    END IF;\n                    RETURN OLD;\n                ",
+                    hash="3d55e73e6ae7d089b5ff136ac16f0c2675285bf2",
+                    operation="DELETE",
+                    pgid="pgtrigger_default_policy_when_auto_policy_a1fd2",
+                    table="desecapi_tokendomainpolicy",
+                    when="BEFORE",
+                ),
+            ),
+        ),
+    ]

api/desecapi/models/tokens.py

@@ -11,7 +11,7 @@
 from django.contrib.postgres.fields import ArrayField
 from django.core import validators
 from django.core.exceptions import ValidationError
-from django.db import models
+from django.db import models, transaction
 from django.db.models import F, Q
 from django.utils import timezone
 from django_prometheus.models import ExportModelOperationsMixin
@@ -52,6 +52,7 @@ def _allowed_subnets_default():
         null=True, default=None, validators=_validators
     )
     domain_policies = models.ManyToManyField("Domain", through="TokenDomainPolicy")
+    auto_policy = models.BooleanField(default=False)
 
     plain = None
     objects = NetManager()
@@ -60,6 +61,27 @@ class Meta:
         constraints = [
             models.UniqueConstraint(fields=["id", "user"], name="unique_id_user")
         ]
+        triggers = [
+            # Ensure that a default policy is defined when auto_policy=true
+            pgtrigger.Trigger(
+                name="token_auto_policy",
+                operation=pgtrigger.Update | pgtrigger.Insert,
+                when=pgtrigger.After,
+                timing=pgtrigger.Deferred,
+                func=pgtrigger.Func(
+                    """
+                    IF
+                        NEW.auto_policy = true AND NOT EXISTS(
+                            SELECT * FROM {meta.many_to_many[0].remote_field.through._meta.db_table} WHERE token_id = NEW.id AND domain_id IS NULL AND subname IS NULL AND type IS NULL
+                        )
+                    THEN
+                        RAISE EXCEPTION 'Token auto policy without a default policy is not allowed. (token.id=%s)', NEW.id;
+                    END IF;
+                    RETURN NULL;
+                """
+                ),
+            ),
+        ]
 
     @property
     def is_valid(self):
@@ -135,6 +157,24 @@ def can_safely_delete_domain(self, domain):
         )
         return not forbidden
 
+    def clean(self):
+        if not self.auto_policy:
+            return
+        default_policy = self.get_policy()
+        if default_policy and default_policy.perm_write:
+            raise ValidationError(
+                {"auto_policy": ["Auto policy requires a restrictive default policy."]}
+            )
+
+    @transaction.atomic
+    def save(self, *args, **kwargs):
+        # Do not perform policy checks when only updating fields like last_used
+        if "auto_policy" in kwargs.get("update_fields", ["auto_policy"]):
+            self.clean()
+        super().save(*args, **kwargs)
+        if self.auto_policy and self.get_policy() is None:
+            TokenDomainPolicy(token=self).save()
+
 
 class TokenDomainPolicy(ExportModelOperationsMixin("TokenDomainPolicy"), models.Model):
     id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
@@ -190,6 +230,22 @@ class Meta:
                 """
                 ),
             ),
+            # Ensure default policy when auto_policy is in effect
+            pgtrigger.Trigger(
+                name="default_policy_when_auto_policy",
+                operation=pgtrigger.Delete,
+                when=pgtrigger.Before,
+                func=pgtrigger.Func(
+                    """
+                    IF
+                        OLD.domain_id IS NULL AND OLD.subname IS NULL AND OLD.type IS NULL AND (SELECT auto_policy FROM {fields.token.remote_field.model._meta.db_table} WHERE id = OLD.token_id) = true
+                    THEN
+                        RAISE EXCEPTION 'Cannot delete default policy while auto_policy is in effect. (tokendomainpolicy.id=%s)', OLD.id;
+                    END IF;
+                    RETURN OLD;
+                """
+                ),
+            ),
         ]
 
     @property
@@ -223,6 +279,15 @@ def clean(self):
                         ]
                     }
                 )
+            # Can't relax default policy if auto_policy is in effect
+            if self.perm_write and self.is_default_policy and self.token.auto_policy:
+                raise ValidationError(
+                    {
+                        "perm_write": [
+                            "Must be false when auto_policy is in effect for the token."
+                        ]
+                    }
+                )
 
     def delete(self, *args, **kwargs):
         # Can't delete default policy when others exist
@@ -237,6 +302,15 @@ def delete(self, *args, **kwargs):
                     ]
                 }
             )
+        # Can't delete default policy when auto_policy is in effect
+        if self.is_default_policy and self.token.auto_policy:
+            raise ValidationError(
+                {
+                    "non_field_errors": [
+                        "Can't delete default policy when auto_policy is in effect for the token."
+                    ]
+                }
+            )
         return super().delete(*args, **kwargs)
 
     def save(self, *args, **kwargs):

api/desecapi/serializers/tokens.py

@@ -23,6 +23,7 @@ class Meta:
             "perm_delete_domain",
             "perm_manage_tokens",
             "allowed_subnets",
+            "auto_policy",
             "is_valid",
             "token",
         )
@@ -38,6 +39,12 @@ def get_fields(self):
             fields.pop("token")
         return fields
 
+    def save(self, **kwargs):
+        try:
+            return super().save(**kwargs)
+        except django.core.exceptions.ValidationError as exc:
+            raise serializers.ValidationError(exc.message_dict)
+
 
 class DomainSlugRelatedField(serializers.SlugRelatedField):
     def get_queryset(self):

api/desecapi/tests/test_token_domain_policy.py

@@ -1,4 +1,4 @@
-from django.db import transaction
+from django.db import connection, transaction
 from django.db.utils import IntegrityError
 from rest_framework import status
 from rest_framework.test import APIClient
@@ -632,3 +632,92 @@ def test_user_deletion(self):
 
         self.token.user.delete()
         self.assertFalse(models.TokenDomainPolicy.objects.filter(pk=policy_pk).exists())
+
+
+class TokenAutoPolicyTestCase(DomainOwnerTestCase):
+    client_class = TokenDomainPolicyClient
+
+    def setUp(self):
+        super().setUp()
+        self.client.credentials()  # remove default credential (corresponding to domain owner)
+        self.token_manage = self.create_token(self.owner, perm_manage_tokens=True)
+        self.other_token = self.create_token(self.user)
+
+    def test_default_policy_constraints(self):
+        self.assertFalse(self.token.tokendomainpolicy_set.exists())
+
+        # Restrictive default policy created when setting auto_policy=true
+        url = DomainOwnerTestCase.reverse("v1:token-detail", pk=self.token.id)
+        response = self.client._request(
+            self.client.patch, url, using=self.token_manage, data={"auto_policy": True}
+        )
+        self.assertStatus(response, status.HTTP_200_OK)
+        self.assertEqual(self.token.tokendomainpolicy_set.count(), 1)
+        default_policy = self.token.get_policy()
+        self.assertFalse(default_policy.perm_write)
+
+        # Can't relax default policy
+        response = self.client.patch_policy(
+            self.token,
+            using=self.token_manage,
+            policy_id=default_policy.id,
+            data={"perm_write": True},
+        )
+        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(
+            response.data["perm_write"][0],
+            "Must be false when auto_policy is in effect for the token.",
+        )
+
+        # Can't delete default policy
+        response = self.client.delete_policy(
+            self.token, using=self.token_manage, policy_id=default_policy.id
+        )
+        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(
+            response.data["non_field_errors"][0],
+            "Can't delete default policy when auto_policy is in effect for the token.",
+        )
+
+        # Can relax default policy when auto_policy=false
+        self.token.auto_policy = False
+        self.token.save()
+        connection.check_constraints()  # simulate transaction commit
+
+        response = self.client.patch_policy(
+            self.token,
+            using=self.token_manage,
+            policy_id=default_policy.id,
+            data={"perm_write": True},
+        )
+        self.assertStatus(response, status.HTTP_200_OK)
+        connection.check_constraints()  # simulate transaction commit
+
+        # Can't set auto_policy when default policy is permissive
+        response = self.client._request(
+            self.client.patch, url, using=self.token_manage, data={"auto_policy": True}
+        )
+        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(
+            response.data["auto_policy"][0],
+            "Auto policy requires a restrictive default policy.",
+        )
+
+        # Can delete default policy when auto_policy=false
+        response = self.client.delete_policy(
+            self.token, using=self.token_manage, policy_id=default_policy.id
+        )
+        self.assertStatus(response, status.HTTP_204_NO_CONTENT)
+
+    def test_auto_policy_from_creation(self):
+        url = DomainOwnerTestCase.reverse("v1:token-list")
+        response = self.client._request(
+            self.client.post, url, using=self.token_manage, data={"auto_policy": True}
+        )
+        self.assertStatus(response, status.HTTP_201_CREATED)
+        self.assertTrue(response.data["auto_policy"])
+
+        # Check that restrictive default policy has been created
+        token = models.Token.objects.get(pk=response.data["id"])
+        self.assertEqual(token.tokendomainpolicy_set.count(), 1)
+        self.assertFalse(token.get_policy().perm_write)

api/desecapi/tests/test_tokens.py

@@ -63,6 +63,7 @@ def test_retrieve_my_token(self):
                 "perm_delete_domain",
                 "perm_manage_tokens",
                 "allowed_subnets",
+                "auto_policy",
                 "is_valid",
             },
         )
@@ -134,6 +135,7 @@ def test_create_token(self):
                     "perm_delete_domain",
                     "perm_manage_tokens",
                     "allowed_subnets",
+                    "auto_policy",
                     "is_valid",
                     "token",
                 },