feat(api): add Token.perm_manage_tokens and surrounding functionality

The migration sets `perm_manage_tokens = True` for all existing tokens,
and then sets the default to `False`.

Related: #347

api/desecapi/migrations/0008_token_perm_manage_tokens.py

@@ -0,0 +1,23 @@
+# Generated by Django 3.1.2 on 2020-11-06 23:02
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('desecapi', '0007_email_citext'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='token',
+            name='perm_manage_tokens',
+            field=models.BooleanField(default=True),
+        ),
+        migrations.AlterField(
+            model_name='token',
+            name='perm_manage_tokens',
+            field=models.BooleanField(default=False),
+        ),
+    ]

api/desecapi/models.py

@@ -197,6 +197,8 @@ class Token(ExportModelOperationsMixin('Token'), rest_framework.authtoken.models
     )
     name = models.CharField('Name', blank=True, max_length=64)
     last_used = models.DateTimeField(null=True, blank=True)
+    perm_manage_tokens = models.BooleanField(default=False)
+
     plain = None
 
     def generate_key(self):

api/desecapi/permissions.py

@@ -32,6 +32,12 @@ def has_permission(self, request, view):
         return ip in IPv4Network('10.8.0.0/24')
 
 
+class ManageTokensPermission(permissions.BasePermission):
+
+    def has_permission(self, request, view):
+        return request.auth.perm_manage_tokens
+
+
 class WithinDomainLimitOnPOST(permissions.BasePermission):
     """
     Permission that requires that the user still has domain limit quota available, if the request is using POST.

api/desecapi/serializers.py

@@ -52,7 +52,7 @@ class TokenSerializer(serializers.ModelSerializer):
 
     class Meta:
         model = models.Token
-        fields = ('id', 'created', 'last_used', 'name', 'token',)
+        fields = ('id', 'created', 'last_used', 'name', 'perm_manage_tokens', 'token',)
         read_only_fields = ('id', 'created', 'last_used', 'token')
 
     def __init__(self, *args, include_plain=False, **kwargs):

api/desecapi/tests/test_tokens.py

@@ -4,10 +4,12 @@
 from desecapi.tests.base import DomainOwnerTestCase
 
 
-class TokenTestCase(DomainOwnerTestCase):
+class TokenPermittedTestCase(DomainOwnerTestCase):
 
     def setUp(self):
         super().setUp()
+        self.token.perm_manage_tokens = True
+        self.token.save()
         self.token2 = self.create_token(self.owner, name='testtoken')
         self.other_token = self.create_token(self.user)
 
@@ -41,7 +43,8 @@ def test_retrieve_my_token(self):
 
         response = self.client.get(url)
         self.assertStatus(response, status.HTTP_200_OK)
-        self.assertTrue(all(field in response.data for field in ['created', 'id', 'last_used', 'name']))
+        self.assertTrue(all(field in response.data for field in ['created', 'id', 'last_used', 'name',
+                                                                 'perm_manage_tokens']))
         self.assertFalse(any(field in response.data for field in ['token', 'key', 'value']))
 
     def test_retrieve_other_token(self):
@@ -57,7 +60,15 @@ def test_update_my_token(self):
         for method in [self.client.patch, self.client.put]:
             response = method(url, data={'name': method.__name__})
             self.assertStatus(response, status.HTTP_200_OK)
-            self.assertEqual(response.data['name'], method.__name__)
+            self.assertEqual(Token.objects.get(pk=self.token.id).name, method.__name__)
+
+        # Revoke token management permission
+        response = self.client.patch(url, data={'perm_manage_tokens': False})
+        self.assertStatus(response, status.HTTP_200_OK)
+
+        # Verify that the change cannot be undone
+        response = self.client.patch(url, data={'perm_manage_tokens': True})
+        self.assertStatus(response, status.HTTP_403_FORBIDDEN)
 
     def test_create_token(self):
         n = len(Token.objects.filter(user=self.owner).all())
@@ -71,3 +82,50 @@ def test_create_token(self):
             self.assertIsNone(response.data['last_used'])
 
         self.assertEqual(len(Token.objects.filter(user=self.owner).all()), n + len(datas))
+
+
+class TokenForbiddenTestCase(DomainOwnerTestCase):
+
+    def setUp(self):
+        super().setUp()
+        self.token2 = self.create_token(self.owner, name='testtoken')
+        self.other_token = self.create_token(self.user)
+
+    def test_token_last_used(self):
+        self.assertIsNone(Token.objects.get(pk=self.token.id).last_used)
+        self.client.get(self.reverse('v1:root'))
+        self.assertIsNotNone(Token.objects.get(pk=self.token.id).last_used)
+
+    def test_list_tokens(self):
+        response = self.client.get(self.reverse('v1:token-list'))
+        self.assertStatus(response, status.HTTP_403_FORBIDDEN)
+
+    def test_delete_my_token(self):
+        for token_id in [Token.objects.get(user=self.owner, name='testtoken').id, self.token.id]:
+            url = self.reverse('v1:token-detail', pk=token_id)
+            response = self.client.delete(url)
+            self.assertStatus(response, status.HTTP_403_FORBIDDEN)
+
+    def test_retrieve_my_token(self):
+        for token_id in [Token.objects.get(user=self.owner, name='testtoken').id, self.token.id]:
+            url = self.reverse('v1:token-detail', pk=token_id)
+            response = self.client.get(url)
+            self.assertStatus(response, status.HTTP_403_FORBIDDEN)
+
+    def test_retrieve_other_token(self):
+        token_id = Token.objects.get(user=self.user).id
+        url = self.reverse('v1:token-detail', pk=token_id)
+        response = self.client.get(url)
+        self.assertStatus(response, status.HTTP_403_FORBIDDEN)
+
+    def test_update_my_token(self):
+        url = self.reverse('v1:token-detail', pk=self.token.id)
+        for method in [self.client.patch, self.client.put]:
+            response = method(url, data={'name': method.__name__})
+            self.assertStatus(response, status.HTTP_403_FORBIDDEN)
+
+    def test_create_token(self):
+        datas = [{}, {'name': ''}, {'name': 'foobar'}]
+        for data in datas:
+            response = self.client.post(self.reverse('v1:token-list'), data=data)
+            self.assertStatus(response, status.HTTP_403_FORBIDDEN)

api/desecapi/views.py

@@ -24,7 +24,7 @@
 from desecapi.exceptions import ConcurrencyException
 from desecapi.pdns import get_serials
 from desecapi.pdns_change_tracker import PDNSChangeTracker
-from desecapi.permissions import IsDomainOwner, IsOwner, IsVPNClient, WithinDomainLimitOnPOST
+from desecapi.permissions import ManageTokensPermission, IsDomainOwner, IsOwner, IsVPNClient, WithinDomainLimitOnPOST
 from desecapi.renderers import PlainTextRenderer
 
 
@@ -78,7 +78,7 @@ def initial(self, request, *args, **kwargs):
 
 class TokenViewSet(IdempotentDestroyMixin, viewsets.ModelViewSet):
     serializer_class = serializers.TokenSerializer
-    permission_classes = (IsAuthenticated,)
+    permission_classes = (IsAuthenticated, ManageTokensPermission,)
     throttle_scope = 'account_management_passive'
 
     def get_queryset(self):
@@ -510,7 +510,7 @@ class AccountLoginView(generics.GenericAPIView):
     def post(self, request, *args, **kwargs):
         user = self.request.user
 
-        token = models.Token.objects.create(user=user, name="login")
+        token = models.Token.objects.create(user=user, name="login", perm_manage_tokens=True)
         user_logged_in.send(sender=user.__class__, request=self.request, user=user)
 
         data = serializers.TokenSerializer(token, include_plain=True).data