authentik.endpoints.api.devices 

  1from datetime import timedelta
  2
  3from django.db.models import OuterRef, Subquery
  4from django.utils.timezone import now
  5from drf_spectacular.utils import extend_schema
  6from rest_framework import mixins
  7from rest_framework.decorators import action
  8from rest_framework.fields import IntegerField, SerializerMethodField
  9from rest_framework.request import Request
 10from rest_framework.response import Response
 11from rest_framework.viewsets import GenericViewSet
 12
 13from authentik.core.api.used_by import UsedByMixin
 14from authentik.core.api.utils import ModelSerializer, PassiveSerializer
 15from authentik.endpoints.api.device_access_group import DeviceAccessGroupSerializer
 16from authentik.endpoints.api.device_connections import DeviceConnectionSerializer
 17from authentik.endpoints.api.device_fact_snapshots import DeviceFactSnapshotSerializer
 18from authentik.endpoints.models import Device, DeviceFactSnapshot
 19
 20
 21class EndpointDeviceSerializer(ModelSerializer):
 22
 23    access_group_obj = DeviceAccessGroupSerializer(source="access_group", required=False)
 24
 25    facts = SerializerMethodField()
 26
 27    def get_facts(self, instance: Device) -> DeviceFactSnapshotSerializer:
 28        return DeviceFactSnapshotSerializer(instance.cached_facts).data
 29
 30    class Meta:
 31        model = Device
 32        fields = [
 33            "device_uuid",
 34            "pbm_uuid",
 35            "name",
 36            "access_group",
 37            "access_group_obj",
 38            "expiring",
 39            "expires",
 40            "facts",
 41            "attributes",
 42        ]
 43
 44
 45class EndpointDeviceDetailsSerializer(EndpointDeviceSerializer):
 46
 47    connections_obj = DeviceConnectionSerializer(many=True, source="deviceconnection_set")
 48
 49    def get_facts(self, instance: Device) -> DeviceFactSnapshotSerializer:
 50        return DeviceFactSnapshotSerializer(instance.facts).data
 51
 52    class Meta(EndpointDeviceSerializer.Meta):
 53        fields = EndpointDeviceSerializer.Meta.fields + [
 54            "connections_obj",
 55            "policies",
 56            "connections",
 57        ]
 58
 59
 60class DeviceViewSet(
 61    UsedByMixin,
 62    mixins.RetrieveModelMixin,
 63    mixins.UpdateModelMixin,
 64    mixins.DestroyModelMixin,
 65    mixins.ListModelMixin,
 66    GenericViewSet,
 67):
 68
 69    queryset = Device.objects.all().select_related("access_group")
 70    serializer_class = EndpointDeviceSerializer
 71    search_fields = [
 72        "name",
 73        "identifier",
 74    ]
 75    ordering = ["identifier"]
 76    filterset_fields = ["name", "identifier"]
 77
 78    class DeviceSummarySerializer(PassiveSerializer):
 79        """Summary of registered devices"""
 80
 81        total_count = IntegerField()
 82        unreachable_count = IntegerField()
 83        outdated_agent_count = IntegerField()
 84
 85    def get_serializer_class(self):
 86        if self.action == "retrieve":
 87            return EndpointDeviceDetailsSerializer
 88        return super().get_serializer_class()
 89
 90    def get_queryset(self):
 91        if self.action == "retrieve":
 92            return super().get_queryset().prefetch_related("connections")
 93        return super().get_queryset()
 94
 95    @extend_schema(responses={200: DeviceSummarySerializer()})
 96    @action(methods=["GET"], detail=False)
 97    def summary(self, request: Request) -> Response:
 98        delta = now() - timedelta(hours=24)
 99        unreachable = (
100            Device.objects.all()
101            .annotate(
102                latest_snapshot=Subquery(
103                    DeviceFactSnapshot.objects.filter(connection__device=OuterRef("pk"))
104                    .order_by("-created")
105                    .values("created")[:1]
106                )
107            )
108            .filter(latest_snapshot__lte=delta)
109            .distinct()
110            .count()
111        )
112        data = {
113            "total_count": Device.objects.all().count(),
114            "unreachable_count": unreachable,
115            # Currently not supported
116            "outdated_agent_count": 0,
117        }
118        return Response(data)
class EndpointDeviceSerializer(authentik.core.api.utils.ModelSerializer): 
22class EndpointDeviceSerializer(ModelSerializer):
23
24    access_group_obj = DeviceAccessGroupSerializer(source="access_group", required=False)
25
26    facts = SerializerMethodField()
27
28    def get_facts(self, instance: Device) -> DeviceFactSnapshotSerializer:
29        return DeviceFactSnapshotSerializer(instance.cached_facts).data
30
31    class Meta:
32        model = Device
33        fields = [
34            "device_uuid",
35            "pbm_uuid",
36            "name",
37            "access_group",
38            "access_group_obj",
39            "expiring",
40            "expires",
41            "facts",
42            "attributes",
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.

access_group_obj
facts
def get_facts( self, instance: authentik.endpoints.models.Device) -> authentik.endpoints.api.device_fact_snapshots.DeviceFactSnapshotSerializer: 
28    def get_facts(self, instance: Device) -> DeviceFactSnapshotSerializer:
29        return DeviceFactSnapshotSerializer(instance.cached_facts).data
Inherited Members
authentik.core.api.utils.ModelSerializer
serializer_field_mapping
update
class EndpointDeviceSerializer.Meta: 
31    class Meta:
32        model = Device
33        fields = [
34            "device_uuid",
35            "pbm_uuid",
36            "name",
37            "access_group",
38            "access_group_obj",
39            "expiring",
40            "expires",
41            "facts",
42            "attributes",
43        ]
fields = ['device_uuid', 'pbm_uuid', 'name', 'access_group', 'access_group_obj', 'expiring', 'expires', 'facts', 'attributes']
class EndpointDeviceDetailsSerializer(EndpointDeviceSerializer): 
46class EndpointDeviceDetailsSerializer(EndpointDeviceSerializer):
47
48    connections_obj = DeviceConnectionSerializer(many=True, source="deviceconnection_set")
49
50    def get_facts(self, instance: Device) -> DeviceFactSnapshotSerializer:
51        return DeviceFactSnapshotSerializer(instance.facts).data
52
53    class Meta(EndpointDeviceSerializer.Meta):
54        fields = EndpointDeviceSerializer.Meta.fields + [
55            "connections_obj",
56            "policies",
57            "connections",
58        ]

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.

connections_obj
def get_facts( self, instance: authentik.endpoints.models.Device) -> authentik.endpoints.api.device_fact_snapshots.DeviceFactSnapshotSerializer: 
50    def get_facts(self, instance: Device) -> DeviceFactSnapshotSerializer:
51        return DeviceFactSnapshotSerializer(instance.facts).data
Inherited Members
EndpointDeviceSerializer
access_group_obj
facts
authentik.core.api.utils.ModelSerializer
serializer_field_mapping
update
class EndpointDeviceDetailsSerializer.Meta(EndpointDeviceSerializer.Meta): 
53    class Meta(EndpointDeviceSerializer.Meta):
54        fields = EndpointDeviceSerializer.Meta.fields + [
55            "connections_obj",
56            "policies",
57            "connections",
58        ]
fields = ['device_uuid', 'pbm_uuid', 'name', 'access_group', 'access_group_obj', 'expiring', 'expires', 'facts', 'attributes', 'connections_obj', 'policies', 'connections']
Inherited Members
EndpointDeviceSerializer.Meta
model
class DeviceViewSet(authentik.core.api.used_by.UsedByMixin, rest_framework.mixins.RetrieveModelMixin, rest_framework.mixins.UpdateModelMixin, rest_framework.mixins.DestroyModelMixin, rest_framework.mixins.ListModelMixin, rest_framework.viewsets.GenericViewSet): 
 61class DeviceViewSet(
 62    UsedByMixin,
 63    mixins.RetrieveModelMixin,
 64    mixins.UpdateModelMixin,
 65    mixins.DestroyModelMixin,
 66    mixins.ListModelMixin,
 67    GenericViewSet,
 68):
 69
 70    queryset = Device.objects.all().select_related("access_group")
 71    serializer_class = EndpointDeviceSerializer
 72    search_fields = [
 73        "name",
 74        "identifier",
 75    ]
 76    ordering = ["identifier"]
 77    filterset_fields = ["name", "identifier"]
 78
 79    class DeviceSummarySerializer(PassiveSerializer):
 80        """Summary of registered devices"""
 81
 82        total_count = IntegerField()
 83        unreachable_count = IntegerField()
 84        outdated_agent_count = IntegerField()
 85
 86    def get_serializer_class(self):
 87        if self.action == "retrieve":
 88            return EndpointDeviceDetailsSerializer
 89        return super().get_serializer_class()
 90
 91    def get_queryset(self):
 92        if self.action == "retrieve":
 93            return super().get_queryset().prefetch_related("connections")
 94        return super().get_queryset()
 95
 96    @extend_schema(responses={200: DeviceSummarySerializer()})
 97    @action(methods=["GET"], detail=False)
 98    def summary(self, request: Request) -> Response:
 99        delta = now() - timedelta(hours=24)
100        unreachable = (
101            Device.objects.all()
102            .annotate(
103                latest_snapshot=Subquery(
104                    DeviceFactSnapshot.objects.filter(connection__device=OuterRef("pk"))
105                    .order_by("-created")
106                    .values("created")[:1]
107                )
108            )
109            .filter(latest_snapshot__lte=delta)
110            .distinct()
111            .count()
112        )
113        data = {
114            "total_count": Device.objects.all().count(),
115            "unreachable_count": unreachable,
116            # Currently not supported
117            "outdated_agent_count": 0,
118        }
119        return Response(data)

Mixin to add a used_by endpoint to return a list of all objects using this object

queryset = <QuerySet []>
serializer_class = <class 'EndpointDeviceSerializer'>
search_fields = ['name', 'identifier']
ordering = ['identifier']
filterset_fields = ['name', 'identifier']
def get_serializer_class(self): 
86    def get_serializer_class(self):
87        if self.action == "retrieve":
88            return EndpointDeviceDetailsSerializer
89        return super().get_serializer_class()

Return the class to use for the serializer. Defaults to using self.serializer_class.

You may want to override this if you need to provide different serializations depending on the incoming request.

(Eg. admins get full serialization, others get basic serialization)

def get_queryset(self): 
91    def get_queryset(self):
92        if self.action == "retrieve":
93            return super().get_queryset().prefetch_related("connections")
94        return super().get_queryset()

Get the list of items for this view. This must be an iterable, and may be a queryset. Defaults to using self.queryset.

This method should always be used rather than accessing self.queryset directly, as self.queryset gets evaluated only once, and those results are cached for all subsequent requests.

You may want to override this if you need to provide different querysets depending on the incoming request.

(Eg. return a list of items that is specific to the user)

@extend_schema(responses={200: DeviceSummarySerializer()})
@action(methods=['GET'], detail=False)
def summary( self, request: rest_framework.request.Request) -> rest_framework.response.Response: 
 96    @extend_schema(responses={200: DeviceSummarySerializer()})
 97    @action(methods=["GET"], detail=False)
 98    def summary(self, request: Request) -> Response:
 99        delta = now() - timedelta(hours=24)
100        unreachable = (
101            Device.objects.all()
102            .annotate(
103                latest_snapshot=Subquery(
104                    DeviceFactSnapshot.objects.filter(connection__device=OuterRef("pk"))
105                    .order_by("-created")
106                    .values("created")[:1]
107                )
108            )
109            .filter(latest_snapshot__lte=delta)
110            .distinct()
111            .count()
112        )
113        data = {
114            "total_count": Device.objects.all().count(),
115            "unreachable_count": unreachable,
116            # Currently not supported
117            "outdated_agent_count": 0,
118        }
119        return Response(data)
name = None
description = None
suffix = None
detail = None
basename = None
Inherited Members
authentik.core.api.used_by.UsedByMixin
used_by
class DeviceViewSet.DeviceSummarySerializer(authentik.core.api.utils.PassiveSerializer): 
79    class DeviceSummarySerializer(PassiveSerializer):
80        """Summary of registered devices"""
81
82        total_count = IntegerField()
83        unreachable_count = IntegerField()
84        outdated_agent_count = IntegerField()

Summary of registered devices

total_count
unreachable_count
outdated_agent_count
Inherited Members
authentik.core.api.utils.PassiveSerializer
create
update