wip feat(api,webapp): add token domain policy and GUI functionality

Related: #347

api/desecapi/migrations/0016_tokendomainpolicy.py

@@ -0,0 +1,43 @@
+# Generated by Django 3.1.3 on 2020-11-23 19:39
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('desecapi', '0015_rrset_touched_index'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='token',
+            name='user',
+            field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL),
+        ),
+        migrations.CreateModel(
+            name='TokenDomainPolicy',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('perm_dyndns', models.BooleanField(default=False)),
+                ('perm_other', models.BooleanField(default=False)),
+                ('domain', models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, to='desecapi.domain')),
+                ('token', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='desecapi.token')),
+            ],
+        ),
+        migrations.AddField(
+            model_name='token',
+            name='domain_policies',
+            field=models.ManyToManyField(through='desecapi.TokenDomainPolicy', to='desecapi.Domain'),
+        ),
+        migrations.AddConstraint(
+            model_name='tokendomainpolicy',
+            constraint=models.UniqueConstraint(fields=('token', 'domain'), name='unique_entry'),
+        ),
+        migrations.AddConstraint(
+            model_name='tokendomainpolicy',
+            constraint=models.UniqueConstraint(condition=models.Q(domain__isnull=True), fields=('token',), name='unique_entry_null_domain'),
+        ),
+    ]

api/desecapi/models.py

@@ -414,16 +414,14 @@ def _allowed_subnets_default():
 
     id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
     key = models.CharField("Key", max_length=128, db_index=True, unique=True)
-    user = models.ForeignKey(
-        User, related_name='auth_tokens',
-        on_delete=models.CASCADE, verbose_name="User"
-    )
+    user = models.ForeignKey(User, on_delete=models.CASCADE)
     name = models.CharField('Name', blank=True, max_length=64)
     last_used = models.DateTimeField(null=True, blank=True)
     perm_manage_tokens = models.BooleanField(default=False)
     allowed_subnets = ArrayField(CidrAddressField(), default=_allowed_subnets_default.__func__)
     max_age = models.DurationField(null=True, default=None, validators=[MinValueValidator(timedelta(0))])
     max_unused_period = models.DurationField(null=True, default=None, validators=[MinValueValidator(timedelta(0))])
+    domain_policies = models.ManyToManyField(Domain, through='TokenDomainPolicy')
 
     plain = None
     objects = NetManager()
@@ -458,6 +456,22 @@ def make_hash(plain):
         return make_password(plain, salt='static', hasher='pbkdf2_sha256_iter1')
 
 
+class TokenDomainPolicy(ExportModelOperationsMixin('TokenDomainPolicy'), models.Model):
+    token = models.ForeignKey(Token, on_delete=models.CASCADE)
+    domain = models.ForeignKey(Domain, on_delete=models.CASCADE, null=True)
+    perm_dyndns = models.BooleanField(default=False)
+    perm_other = models.BooleanField(default=False)
+
+    class Meta:
+        constraints = [
+            models.UniqueConstraint(fields=['token', 'domain'], name='unique_entry'),
+            models.UniqueConstraint(fields=['token'], condition=Q(domain__isnull=True), name='unique_entry_null_domain')
+        ]
+
+    @property
+    def any_perm(self):
+        return any(getattr(self, field.name) for field in self._meta.get_fields() if field.name.startswith('perm_'))
+
 class Donation(ExportModelOperationsMixin('Donation'), models.Model):
     @staticmethod
     def _created_default():

api/desecapi/permissions.py

@@ -2,6 +2,8 @@
 
 from rest_framework import permissions
 
+from desecapi.models import TokenDomainPolicy
+
 
 class IsOwner(permissions.BasePermission):
     """
@@ -21,6 +23,48 @@ def has_object_permission(self, request, view, obj):
         return obj.domain.owner == request.user
 
 
+class TokenDomainPolicyBasePermission(permissions.BasePermission):
+    """
+    Base permission to check whether a token authorizes specific actions on a domain.
+    """
+    perm_field = 'any_perm'
+
+    def _has_object_permission(self, request, view, obj):
+        ### TODO reduce number of queries?
+
+        # Try domain-specific policy first
+        try:
+            return getattr(TokenDomainPolicy.objects.get(token=request.auth, domain=obj), self.perm_field)
+        except TokenDomainPolicy.DoesNotExist:
+            pass
+
+        # Try general policy
+        try:
+            return getattr(TokenDomainPolicy.objects.get(token=request.auth, domain__isnull=True), self.perm_field)
+        except TokenDomainPolicy.DoesNotExist:
+            pass
+
+        # Else, allow if and only if the token has no domain policy at all
+        return not TokenDomainPolicy.objects.filter(token=request.auth).exists()
+
+
+class TokenHasDomainObjectPermission(TokenDomainPolicyBasePermission):
+    has_object_permission = TokenDomainPolicyBasePermission._has_object_permission
+
+
+class TokenHasViewDomainPermission(TokenDomainPolicyBasePermission):
+
+    def has_permission(self, request, view):
+        return self._has_object_permission(request, view, view.domain)
+
+
+class TokenHasViewDomainDynPermission(TokenHasViewDomainPermission):
+    """
+    Custom permission to check whether a token authorizes using the dynDNS interface for the view domain.
+    """
+    perm_field = 'perm_dyndns'
+
+
 class IsVPNClient(permissions.BasePermission):
     """
     Permission that requires that the user is accessing using an IP from the VPN net.

api/desecapi/serializers.py

@@ -75,6 +75,14 @@ def get_fields(self):
         return fields
 
 
+class TokenDomainPolicySerializer(serializers.ModelSerializer):
+
+    class Meta:
+        model = models.TokenDomainPolicy
+        fields = ('domain', 'perm_dyndns', 'perm_other',)
+        read_only_fields = ('domain',)
+
+
 class RequiredOnPartialUpdateCharField(serializers.CharField):
     """
     This field is always required, even for partial updates (e.g. using PATCH).

api/desecapi/tests/test_token_domain_policy.py

@@ -0,0 +1,18 @@
+from rest_framework import status
+
+from desecapi.models import Token
+from desecapi.tests.base import DomainOwnerTestCase
+
+
+class TokenDomainPolicyTestCase(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)
+
+    def test_list_domains(self):
+        response = self.client.get(self.reverse('v1:domain-list'))
+        self.assertStatus(response, status.HTTP_200_OK)

api/desecapi/tests/test_tokens.py

@@ -1,5 +1,3 @@
-from ipaddress import IPv4Network
-
 from rest_framework import status
 
 from desecapi.models import Token

api/desecapi/urls/version_1.py

@@ -6,6 +6,9 @@
 tokens_router = SimpleRouter()
 tokens_router.register(r'', views.TokenViewSet, basename='token')
 
+tokendomainpolicies_router = SimpleRouter()
+tokendomainpolicies_router.register(r'', views.TokenDomainPolicyViewSet, basename='token_domain_policies')
+
 auth_urls = [
     # User management
     path('', views.AccountCreateView.as_view(), name='register'),
@@ -18,6 +21,7 @@
 
     # Token management
     path('tokens/', include(tokens_router.urls)),
+    path('tokens/<id>/domain_policies/', include(tokendomainpolicies_router.urls))
 ]
 
 domains_router = SimpleRouter()

api/desecapi/views.py

@@ -25,7 +25,10 @@
 from desecapi.exceptions import ConcurrencyException
 from desecapi.pdns import get_serials
 from desecapi.pdns_change_tracker import PDNSChangeTracker
-from desecapi.permissions import ManageTokensPermission, IsDomainOwner, IsOwner, IsVPNClient, WithinDomainLimitOnPOST
+from desecapi.permissions import (
+    IsDomainOwner, IsOwner, IsVPNClient, ManageTokensPermission, TokenHasDomainObjectPermission,
+    TokenHasViewDomainDynPermission, TokenHasViewDomainPermission, WithinDomainLimitOnPOST,
+)
 from desecapi.renderers import PlainTextRenderer
 
 
@@ -67,26 +70,27 @@ def destroy(self, request, *args, **kwargs):
 
 class DomainViewMixin:
 
-    def get_serializer_context(self):
-        return {**super().get_serializer_context(), 'domain': self.domain}
-
-    def initial(self, request, *args, **kwargs):
-        # noinspection PyUnresolvedReferences
-        super().initial(request, *args, **kwargs)
+    @property
+    def domain(self):
         try:
-            # noinspection PyAttributeOutsideInit, PyUnresolvedReferences
-            self.domain = self.request.user.domains.get(name=self.kwargs['name'])
+            return self.request.user.domains.get(name=self.kwargs['name'])
         except models.Domain.DoesNotExist:
             raise Http404
 
+    def get_permissions(self):
+        return [TokenHasViewDomainPermission()] + super().get_permissions()
+
+    def get_serializer_context(self):
+        return {**super().get_serializer_context(), 'domain': self.domain}
+
 
 class TokenViewSet(IdempotentDestroyMixin, viewsets.ModelViewSet):
     serializer_class = serializers.TokenSerializer
     permission_classes = (IsAuthenticated, ManageTokensPermission,)
     throttle_scope = 'account_management_passive'
 
     def get_queryset(self):
-        return self.request.user.auth_tokens.all()
+        return self.request.user.token_set.all()
 
     def get_serializer(self, *args, **kwargs):
         # When creating a new token, return the plaintext representation
@@ -98,14 +102,25 @@ def perform_create(self, serializer):
         serializer.save(user=self.request.user)
 
 
+class TokenDomainPolicyViewSet(viewsets.ModelViewSet):
+    serializer_class = serializers.TokenDomainPolicySerializer
+    permission_classes = (IsAuthenticated,)
+    throttle_scope = 'account_management_passive'
+
+    def get_queryset(self):
+        ### TODO token manage permission?
+        print('============= ID', self.kwargs['id'])
+        return self.request.user.token_set.get(id=self.kwargs['id']).domain_policies
+
+
 class DomainViewSet(IdempotentDestroyMixin,
                     mixins.CreateModelMixin,
                     mixins.RetrieveModelMixin,
                     mixins.DestroyModelMixin,
                     mixins.ListModelMixin,
                     viewsets.GenericViewSet):
     serializer_class = serializers.DomainSerializer
-    permission_classes = (IsAuthenticated, IsOwner, WithinDomainLimitOnPOST)
+    permission_classes = (IsAuthenticated, IsOwner, WithinDomainLimitOnPOST, TokenHasDomainObjectPermission)
     lookup_field = 'name'
     lookup_value_regex = r'[^/]+'
 
@@ -221,6 +236,7 @@ def get_object(self):
         # is fine as per https://www.django-rest-framework.org/api-guide/serializers/#serializing-multiple-objects.
         # We skip checking object permissions here to avoid evaluating the queryset. The user can access all his RRsets
         # anyways.
+        ### TODO include permission check
         return self.filter_queryset(self.get_queryset())
 
     def get_serializer(self, *args, **kwargs):
@@ -268,6 +284,7 @@ def get(self, request, *_):
 
 class DynDNS12Update(generics.GenericAPIView):
     authentication_classes = (auth.TokenAuthentication, auth.BasicTokenAuthentication, auth.URLParamAuthentication,)
+    permission_classes = [TokenHasViewDomainDynPermission]
     renderer_classes = [PlainTextRenderer]
     throttle_scope = 'dyndns'