authentik.providers.oauth2.api.tokens

  1"""OAuth2Provider API Views"""
  2
  3from json import dumps
  4
  5from django_filters.rest_framework import DjangoFilterBackend
  6from guardian.shortcuts import get_anonymous_user
  7from rest_framework import mixins
  8from rest_framework.fields import CharField, ListField, SerializerMethodField
  9from rest_framework.filters import OrderingFilter, SearchFilter
 10from rest_framework.viewsets import GenericViewSet
 11
 12from authentik.core.api.used_by import UsedByMixin
 13from authentik.core.api.users import UserSerializer
 14from authentik.core.api.utils import MetaNameSerializer, ModelSerializer
 15from authentik.providers.oauth2.api.providers import OAuth2ProviderSerializer
 16from authentik.providers.oauth2.models import AccessToken, AuthorizationCode, RefreshToken
 17
 18
 19class ExpiringBaseGrantModelSerializer(ModelSerializer, MetaNameSerializer):
 20    """Serializer for BaseGrantModel and ExpiringBaseGrant"""
 21
 22    user = UserSerializer()
 23    provider = OAuth2ProviderSerializer()
 24    scope = ListField(child=CharField())
 25
 26    class Meta:
 27        model = AuthorizationCode
 28        fields = ["pk", "provider", "user", "is_expired", "expires", "scope"]
 29        depth = 2
 30
 31
 32class TokenModelSerializer(ExpiringBaseGrantModelSerializer):
 33    """Serializer for BaseGrantModel and RefreshToken"""
 34
 35    id_token = SerializerMethodField()
 36
 37    def get_id_token(self, instance: RefreshToken) -> str:
 38        """Get the token's id_token as JSON String"""
 39        return dumps(instance.id_token.to_dict(), indent=4)
 40
 41    class Meta:
 42        model = RefreshToken
 43        fields = [
 44            "pk",
 45            "provider",
 46            "user",
 47            "is_expired",
 48            "expires",
 49            "scope",
 50            "id_token",
 51            "revoked",
 52        ]
 53        depth = 2
 54
 55
 56class AuthorizationCodeViewSet(
 57    mixins.RetrieveModelMixin,
 58    mixins.DestroyModelMixin,
 59    UsedByMixin,
 60    mixins.ListModelMixin,
 61    GenericViewSet,
 62):
 63    """AuthorizationCode Viewset"""
 64
 65    queryset = AuthorizationCode.objects.including_expired().all()
 66    serializer_class = ExpiringBaseGrantModelSerializer
 67    filterset_fields = ["user", "provider"]
 68    ordering = ["provider", "expires"]
 69    filter_backends = [
 70        DjangoFilterBackend,
 71        OrderingFilter,
 72        SearchFilter,
 73    ]
 74
 75    def get_queryset(self):
 76        user = self.request.user if self.request else get_anonymous_user()
 77        if user.is_superuser:
 78            return super().get_queryset()
 79        return super().get_queryset().filter(user=user.pk)
 80
 81
 82class RefreshTokenViewSet(
 83    mixins.RetrieveModelMixin,
 84    mixins.DestroyModelMixin,
 85    UsedByMixin,
 86    mixins.ListModelMixin,
 87    GenericViewSet,
 88):
 89    """RefreshToken Viewset"""
 90
 91    queryset = RefreshToken.objects.including_expired().all()
 92    serializer_class = TokenModelSerializer
 93    filterset_fields = ["user", "provider"]
 94    ordering = ["provider", "expires"]
 95    filter_backends = [
 96        DjangoFilterBackend,
 97        OrderingFilter,
 98        SearchFilter,
 99    ]
100
101    def get_queryset(self):
102        user = self.request.user if self.request else get_anonymous_user()
103        if user.is_superuser:
104            return super().get_queryset()
105        return super().get_queryset().filter(user=user.pk)
106
107
108class AccessTokenViewSet(
109    mixins.RetrieveModelMixin,
110    mixins.DestroyModelMixin,
111    UsedByMixin,
112    mixins.ListModelMixin,
113    GenericViewSet,
114):
115    """AccessToken Viewset"""
116
117    queryset = AccessToken.objects.including_expired().all()
118    serializer_class = TokenModelSerializer
119    filterset_fields = ["user", "provider"]
120    ordering = ["provider", "expires"]
121    filter_backends = [
122        DjangoFilterBackend,
123        OrderingFilter,
124        SearchFilter,
125    ]
126
127    def get_queryset(self):
128        user = self.request.user if self.request else get_anonymous_user()
129        if user.is_superuser:
130            return super().get_queryset()
131        return super().get_queryset().filter(user=user.pk)