authentik.endpoints.connectors.agent.api.enrollment_tokens
1from drf_spectacular.utils import OpenApiResponse, extend_schema 2from rest_framework.decorators import action 3from rest_framework.fields import CharField 4from rest_framework.request import Request 5from rest_framework.response import Response 6from rest_framework.viewsets import ModelViewSet 7 8from authentik.blueprints.v1.importer import SERIALIZER_CONTEXT_BLUEPRINT 9from authentik.core.api.tokens import TokenViewSerializer 10from authentik.core.api.used_by import UsedByMixin 11from authentik.core.api.utils import ModelSerializer 12from authentik.endpoints.api.device_access_group import DeviceAccessGroupSerializer 13from authentik.endpoints.connectors.agent.models import EnrollmentToken 14from authentik.events.models import Event, EventAction 15from authentik.rbac.decorators import permission_required 16 17 18class EnrollmentTokenSerializer(ModelSerializer): 19 20 device_group_obj = DeviceAccessGroupSerializer( 21 source="device_group", 22 read_only=True, 23 required=False, 24 allow_null=True, 25 ) 26 27 def __init__(self, *args, **kwargs) -> None: 28 super().__init__(*args, **kwargs) 29 if SERIALIZER_CONTEXT_BLUEPRINT in self.context: 30 self.fields["key"] = CharField(required=False) 31 32 class Meta: 33 model = EnrollmentToken 34 fields = [ 35 "token_uuid", 36 "device_group", 37 "device_group_obj", 38 "connector", 39 "name", 40 "expiring", 41 "expires", 42 ] 43 44 45class EnrollmentTokenViewSet(UsedByMixin, ModelViewSet): 46 47 queryset = EnrollmentToken.objects.all().prefetch_related("device_group") 48 serializer_class = EnrollmentTokenSerializer 49 search_fields = [ 50 "name", 51 "connector__name", 52 ] 53 ordering = ["token_uuid"] 54 filterset_fields = ["token_uuid", "connector"] 55 56 @permission_required("authentik_endpoints_connectors_agent.view_enrollment_token_key") 57 @extend_schema( 58 responses={ 59 200: TokenViewSerializer(many=False), 60 404: OpenApiResponse(description="Token not found or expired"), 61 } 62 ) 63 @action(detail=True, pagination_class=None, filter_backends=[], methods=["GET"]) 64 def view_key(self, request: Request, pk: str) -> Response: 65 """Return token key and log access""" 66 token: EnrollmentToken = self.get_object() 67 Event.new(EventAction.SECRET_VIEW, secret=token).from_http(request) # noqa # nosec 68 return Response(TokenViewSerializer({"key": token.key}).data)
19class EnrollmentTokenSerializer(ModelSerializer): 20 21 device_group_obj = DeviceAccessGroupSerializer( 22 source="device_group", 23 read_only=True, 24 required=False, 25 allow_null=True, 26 ) 27 28 def __init__(self, *args, **kwargs) -> None: 29 super().__init__(*args, **kwargs) 30 if SERIALIZER_CONTEXT_BLUEPRINT in self.context: 31 self.fields["key"] = CharField(required=False) 32 33 class Meta: 34 model = EnrollmentToken 35 fields = [ 36 "token_uuid", 37 "device_group", 38 "device_group_obj", 39 "connector", 40 "name", 41 "expiring", 42 "expires", 43 ]
A ModelSerializer is just a regular Serializer, except that:
- A set of default fields are automatically populated.
- A set of default validators are automatically populated.
- Default
.create()and.update()implementations are provided.
The process of automatically determining a set of serializer fields based on the model fields is reasonably complex, but you almost certainly don't need to dig into the implementation.
If the ModelSerializer class doesn't generate the set of fields that
you need you should either declare the extra/differing fields explicitly on
the serializer class, or simply use a Serializer class.
Inherited Members
class
EnrollmentTokenSerializer.Meta:
33 class Meta: 34 model = EnrollmentToken 35 fields = [ 36 "token_uuid", 37 "device_group", 38 "device_group_obj", 39 "connector", 40 "name", 41 "expiring", 42 "expires", 43 ]
model =
<class 'authentik.endpoints.connectors.agent.models.EnrollmentToken'>
class
EnrollmentTokenViewSet(authentik.core.api.used_by.UsedByMixin, rest_framework.viewsets.ModelViewSet):
46class EnrollmentTokenViewSet(UsedByMixin, ModelViewSet): 47 48 queryset = EnrollmentToken.objects.all().prefetch_related("device_group") 49 serializer_class = EnrollmentTokenSerializer 50 search_fields = [ 51 "name", 52 "connector__name", 53 ] 54 ordering = ["token_uuid"] 55 filterset_fields = ["token_uuid", "connector"] 56 57 @permission_required("authentik_endpoints_connectors_agent.view_enrollment_token_key") 58 @extend_schema( 59 responses={ 60 200: TokenViewSerializer(many=False), 61 404: OpenApiResponse(description="Token not found or expired"), 62 } 63 ) 64 @action(detail=True, pagination_class=None, filter_backends=[], methods=["GET"]) 65 def view_key(self, request: Request, pk: str) -> Response: 66 """Return token key and log access""" 67 token: EnrollmentToken = self.get_object() 68 Event.new(EventAction.SECRET_VIEW, secret=token).from_http(request) # noqa # nosec 69 return Response(TokenViewSerializer({"key": token.key}).data)
Mixin to add a used_by endpoint to return a list of all objects using this object
serializer_class =
<class 'EnrollmentTokenSerializer'>
@extend_schema(responses={200: TokenViewSerializer(many=False), 404: OpenApiResponse(description='Token not found or expired')})
@action(detail=True, pagination_class=None, filter_backends=[], methods=['GET'])
def
view_key( self, request: rest_framework.request.Request, pk: str) -> rest_framework.response.Response:
57 @permission_required("authentik_endpoints_connectors_agent.view_enrollment_token_key") 58 @extend_schema( 59 responses={ 60 200: TokenViewSerializer(many=False), 61 404: OpenApiResponse(description="Token not found or expired"), 62 } 63 ) 64 @action(detail=True, pagination_class=None, filter_backends=[], methods=["GET"]) 65 def view_key(self, request: Request, pk: str) -> Response: 66 """Return token key and log access""" 67 token: EnrollmentToken = self.get_object() 68 Event.new(EventAction.SECRET_VIEW, secret=token).from_http(request) # noqa # nosec 69 return Response(TokenViewSerializer({"key": token.key}).data)
Return token key and log access