desec-stack/api/desecapi/views/tokens.py

import django.core.exceptions
from rest_framework import viewsets
from rest_framework.exceptions import ValidationError
from rest_framework.permissions import IsAuthenticated, SAFE_METHODS
from rest_framework.response import Response
from rest_framework.reverse import reverse
from rest_framework.views import APIView

from desecapi import permissions
from desecapi.models import TokenDomainPolicy
from desecapi.serializers import TokenDomainPolicySerializer, TokenSerializer

from .base import IdempotentDestroyMixin
from .domains import DomainViewSet


class TokenViewSet(IdempotentDestroyMixin, viewsets.ModelViewSet):
    serializer_class = TokenSerializer
    permission_classes = (
        IsAuthenticated,
        permissions.HasManageTokensPermission,
    )
    throttle_scope = "account_management_passive"

    def get_queryset(self):
        return self.request.user.token_set.all()

    def get_serializer(self, *args, **kwargs):
        # When creating a new token, return the plaintext representation
        if self.action == "create":
            kwargs.setdefault("include_plain", True)
        return super().get_serializer(*args, **kwargs)

    def perform_create(self, serializer):
        serializer.save(user=self.request.user)


class TokenPoliciesRoot(APIView):
    permission_classes = [
        IsAuthenticated,
        permissions.HasManageTokensPermission
        | permissions.AuthTokenCorrespondsToViewToken,
    ]

    def get(self, request, *args, **kwargs):
        return Response(
            {
                "domain": reverse(
                    "token_domain_policies-list", request=request, kwargs=kwargs
                )
            }
        )


class TokenDomainPolicyViewSet(IdempotentDestroyMixin, viewsets.ModelViewSet):
    lookup_field = "domain__name"
    lookup_value_regex = DomainViewSet.lookup_value_regex
    pagination_class = None
    serializer_class = TokenDomainPolicySerializer
    throttle_scope = "account_management_passive"

    @property
    def permission_classes(self):
        ret = [IsAuthenticated]
        if self.request.method in SAFE_METHODS:
            ret.append(
                permissions.HasManageTokensPermission
                | permissions.AuthTokenCorrespondsToViewToken
            )
        else:
            ret.append(permissions.HasManageTokensPermission)
        return ret

    def dispatch(self, request, *args, **kwargs):
        # map default policy onto domain_id IS NULL
        lookup_url_kwarg = self.lookup_url_kwarg or self.lookup_field
        try:
            if kwargs[lookup_url_kwarg] == "default":
                kwargs[lookup_url_kwarg] = None
        except KeyError:
            pass
        return super().dispatch(request, *args, **kwargs)

    def get_queryset(self):
        return TokenDomainPolicy.objects.filter(
            token_id=self.kwargs["token_id"], token__user=self.request.user
        )

    def perform_destroy(self, instance):
        try:
            super().perform_destroy(instance)
        except django.core.exceptions.ValidationError as exc:
            raise ValidationError(exc.message_dict, code="precedence")