authentik.core.api.sources

Source API Views

  1"""Source API Views"""
  2
  3from collections.abc import Iterable
  4
  5from drf_spectacular.utils import extend_schema
  6from rest_framework import mixins
  7from rest_framework.decorators import action
  8from rest_framework.exceptions import ValidationError
  9from rest_framework.fields import ReadOnlyField, SerializerMethodField
 10from rest_framework.request import Request
 11from rest_framework.response import Response
 12from rest_framework.viewsets import GenericViewSet
 13from structlog.stdlib import get_logger
 14
 15from authentik.core.api.object_types import TypesMixin
 16from authentik.core.api.used_by import UsedByMixin
 17from authentik.core.api.utils import MetaNameSerializer, ModelSerializer, ThemedUrlsSerializer
 18from authentik.core.models import GroupSourceConnection, Source, UserSourceConnection
 19from authentik.core.types import UserSettingSerializer
 20from authentik.policies.engine import PolicyEngine
 21
 22LOGGER = get_logger()
 23
 24
 25class SourceSerializer(ModelSerializer, MetaNameSerializer):
 26    """Source Serializer"""
 27
 28    managed = ReadOnlyField()
 29    component = SerializerMethodField()
 30    icon_url = ReadOnlyField()
 31    icon_themed_urls = ThemedUrlsSerializer(read_only=True, allow_null=True)
 32
 33    def get_component(self, obj: Source) -> str:
 34        """Get object component so that we know how to edit the object"""
 35        if obj.__class__ == Source:
 36            return ""
 37        return obj.component
 38
 39    class Meta:
 40        model = Source
 41        fields = [
 42            "pk",
 43            "name",
 44            "slug",
 45            "enabled",
 46            "promoted",
 47            "authentication_flow",
 48            "enrollment_flow",
 49            "user_property_mappings",
 50            "group_property_mappings",
 51            "component",
 52            "verbose_name",
 53            "verbose_name_plural",
 54            "meta_model_name",
 55            "policy_engine_mode",
 56            "user_matching_mode",
 57            "managed",
 58            "user_path_template",
 59            "icon",
 60            "icon_url",
 61            "icon_themed_urls",
 62        ]
 63
 64
 65class SourceViewSet(
 66    TypesMixin,
 67    mixins.RetrieveModelMixin,
 68    mixins.DestroyModelMixin,
 69    UsedByMixin,
 70    mixins.ListModelMixin,
 71    GenericViewSet,
 72):
 73    """Source Viewset"""
 74
 75    queryset = Source.objects.none()
 76    serializer_class = SourceSerializer
 77    lookup_field = "slug"
 78    search_fields = ["slug", "name"]
 79    filterset_fields = ["slug", "name", "managed", "pbm_uuid"]
 80
 81    def get_queryset(self):  # pragma: no cover
 82        return Source.objects.select_subclasses()
 83
 84    @extend_schema(responses={200: UserSettingSerializer(many=True)})
 85    @action(detail=False, pagination_class=None, filter_backends=[])
 86    def user_settings(self, request: Request) -> Response:
 87        """Get all sources the user can configure"""
 88        _all_sources: Iterable[Source] = (
 89            Source.objects.filter(enabled=True).select_subclasses().order_by("name")
 90        )
 91        matching_sources: list[UserSettingSerializer] = []
 92        for source in _all_sources:
 93            user_settings = source.ui_user_settings()
 94            if not user_settings:
 95                continue
 96            policy_engine = PolicyEngine(source, request.user, request)
 97            policy_engine.build()
 98            if not policy_engine.passing:
 99                continue
100            source_settings = source.ui_user_settings()
101            source_settings.initial_data["object_uid"] = source.slug
102            if not source_settings.is_valid():
103                LOGGER.warning(source_settings.errors)
104            matching_sources.append(source_settings.validated_data)
105        return Response(matching_sources)
106
107    def destroy(self, request: Request, *args, **kwargs):
108        """Prevent deletion of built-in sources"""
109        instance: Source = self.get_object()
110
111        if instance.managed == Source.MANAGED_INBUILT:
112            raise ValidationError(
113                {"detail": "Built-in sources cannot be deleted"}, code="protected"
114            )
115
116        return super().destroy(request, *args, **kwargs)
117
118
119class UserSourceConnectionSerializer(SourceSerializer):
120    """User source connection"""
121
122    source_obj = SourceSerializer(read_only=True, source="source")
123
124    class Meta:
125        model = UserSourceConnection
126        fields = [
127            "pk",
128            "user",
129            "source",
130            "source_obj",
131            "identifier",
132            "created",
133            "last_updated",
134        ]
135        extra_kwargs = {
136            "created": {"read_only": True},
137            "last_updated": {"read_only": True},
138        }
139
140
141class UserSourceConnectionViewSet(
142    mixins.RetrieveModelMixin,
143    mixins.UpdateModelMixin,
144    mixins.DestroyModelMixin,
145    UsedByMixin,
146    mixins.ListModelMixin,
147    GenericViewSet,
148):
149    """User-source connection Viewset"""
150
151    queryset = UserSourceConnection.objects.all()
152    serializer_class = UserSourceConnectionSerializer
153    filterset_fields = ["user", "source__slug"]
154    search_fields = ["user__username", "source__slug", "identifier"]
155    ordering = ["source__slug", "pk"]
156    owner_field = "user"
157
158
159class GroupSourceConnectionSerializer(SourceSerializer):
160    """Group Source Connection"""
161
162    source_obj = SourceSerializer(read_only=True)
163
164    class Meta:
165        model = GroupSourceConnection
166        fields = [
167            "pk",
168            "group",
169            "source",
170            "source_obj",
171            "identifier",
172            "created",
173            "last_updated",
174        ]
175        extra_kwargs = {
176            "created": {"read_only": True},
177            "last_updated": {"read_only": True},
178        }
179
180
181class GroupSourceConnectionViewSet(
182    mixins.RetrieveModelMixin,
183    mixins.UpdateModelMixin,
184    mixins.DestroyModelMixin,
185    UsedByMixin,
186    mixins.ListModelMixin,
187    GenericViewSet,
188):
189    """Group-source connection Viewset"""
190
191    queryset = GroupSourceConnection.objects.all()
192    serializer_class = GroupSourceConnectionSerializer
193    filterset_fields = ["group", "source__slug"]
194    search_fields = ["group__name", "source__slug", "identifier"]
195    ordering = ["source__slug", "pk"]
LOGGER = <BoundLoggerLazyProxy(logger=None, wrapper_class=None, processors=None, context_class=None, initial_values={}, logger_factory_args=())>
class SourceSerializer(authentik.core.api.utils.ModelSerializer, authentik.core.api.utils.MetaNameSerializer): 
26class SourceSerializer(ModelSerializer, MetaNameSerializer):
27    """Source Serializer"""
28
29    managed = ReadOnlyField()
30    component = SerializerMethodField()
31    icon_url = ReadOnlyField()
32    icon_themed_urls = ThemedUrlsSerializer(read_only=True, allow_null=True)
33
34    def get_component(self, obj: Source) -> str:
35        """Get object component so that we know how to edit the object"""
36        if obj.__class__ == Source:
37            return ""
38        return obj.component
39
40    class Meta:
41        model = Source
42        fields = [
43            "pk",
44            "name",
45            "slug",
46            "enabled",
47            "promoted",
48            "authentication_flow",
49            "enrollment_flow",
50            "user_property_mappings",
51            "group_property_mappings",
52            "component",
53            "verbose_name",
54            "verbose_name_plural",
55            "meta_model_name",
56            "policy_engine_mode",
57            "user_matching_mode",
58            "managed",
59            "user_path_template",
60            "icon",
61            "icon_url",
62            "icon_themed_urls",
63        ]

Source Serializer

managed
component
icon_url
icon_themed_urls
def get_component(self, obj: authentik.core.models.Source) -> str: 
34    def get_component(self, obj: Source) -> str:
35        """Get object component so that we know how to edit the object"""
36        if obj.__class__ == Source:
37            return ""
38        return obj.component

Get object component so that we know how to edit the object

Inherited Members
authentik.core.api.utils.ModelSerializer
serializer_field_mapping
update
authentik.core.api.utils.MetaNameSerializer
verbose_name
verbose_name_plural
meta_model_name
get_verbose_name
get_verbose_name_plural
get_meta_model_name
class SourceSerializer.Meta: 
40    class Meta:
41        model = Source
42        fields = [
43            "pk",
44            "name",
45            "slug",
46            "enabled",
47            "promoted",
48            "authentication_flow",
49            "enrollment_flow",
50            "user_property_mappings",
51            "group_property_mappings",
52            "component",
53            "verbose_name",
54            "verbose_name_plural",
55            "meta_model_name",
56            "policy_engine_mode",
57            "user_matching_mode",
58            "managed",
59            "user_path_template",
60            "icon",
61            "icon_url",
62            "icon_themed_urls",
63        ]
model = <class 'authentik.core.models.Source'>
fields = ['pk', 'name', 'slug', 'enabled', 'promoted', 'authentication_flow', 'enrollment_flow', 'user_property_mappings', 'group_property_mappings', 'component', 'verbose_name', 'verbose_name_plural', 'meta_model_name', 'policy_engine_mode', 'user_matching_mode', 'managed', 'user_path_template', 'icon', 'icon_url', 'icon_themed_urls']
class SourceViewSet(authentik.core.api.object_types.TypesMixin, rest_framework.mixins.RetrieveModelMixin, rest_framework.mixins.DestroyModelMixin, authentik.core.api.used_by.UsedByMixin, rest_framework.mixins.ListModelMixin, rest_framework.viewsets.GenericViewSet): 
 66class SourceViewSet(
 67    TypesMixin,
 68    mixins.RetrieveModelMixin,
 69    mixins.DestroyModelMixin,
 70    UsedByMixin,
 71    mixins.ListModelMixin,
 72    GenericViewSet,
 73):
 74    """Source Viewset"""
 75
 76    queryset = Source.objects.none()
 77    serializer_class = SourceSerializer
 78    lookup_field = "slug"
 79    search_fields = ["slug", "name"]
 80    filterset_fields = ["slug", "name", "managed", "pbm_uuid"]
 81
 82    def get_queryset(self):  # pragma: no cover
 83        return Source.objects.select_subclasses()
 84
 85    @extend_schema(responses={200: UserSettingSerializer(many=True)})
 86    @action(detail=False, pagination_class=None, filter_backends=[])
 87    def user_settings(self, request: Request) -> Response:
 88        """Get all sources the user can configure"""
 89        _all_sources: Iterable[Source] = (
 90            Source.objects.filter(enabled=True).select_subclasses().order_by("name")
 91        )
 92        matching_sources: list[UserSettingSerializer] = []
 93        for source in _all_sources:
 94            user_settings = source.ui_user_settings()
 95            if not user_settings:
 96                continue
 97            policy_engine = PolicyEngine(source, request.user, request)
 98            policy_engine.build()
 99            if not policy_engine.passing:
100                continue
101            source_settings = source.ui_user_settings()
102            source_settings.initial_data["object_uid"] = source.slug
103            if not source_settings.is_valid():
104                LOGGER.warning(source_settings.errors)
105            matching_sources.append(source_settings.validated_data)
106        return Response(matching_sources)
107
108    def destroy(self, request: Request, *args, **kwargs):
109        """Prevent deletion of built-in sources"""
110        instance: Source = self.get_object()
111
112        if instance.managed == Source.MANAGED_INBUILT:
113            raise ValidationError(
114                {"detail": "Built-in sources cannot be deleted"}, code="protected"
115            )
116
117        return super().destroy(request, *args, **kwargs)

Source Viewset

queryset = <InheritanceQuerySet []>
serializer_class = <class 'SourceSerializer'>
lookup_field = 'slug'
search_fields = ['slug', 'name']
filterset_fields = ['slug', 'name', 'managed', 'pbm_uuid']
def get_queryset(self): 
82    def get_queryset(self):  # pragma: no cover
83        return Source.objects.select_subclasses()

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: UserSettingSerializer(many=True)})
@action(detail=False, pagination_class=None, filter_backends=[])
def user_settings( self, request: rest_framework.request.Request) -> rest_framework.response.Response: 
 85    @extend_schema(responses={200: UserSettingSerializer(many=True)})
 86    @action(detail=False, pagination_class=None, filter_backends=[])
 87    def user_settings(self, request: Request) -> Response:
 88        """Get all sources the user can configure"""
 89        _all_sources: Iterable[Source] = (
 90            Source.objects.filter(enabled=True).select_subclasses().order_by("name")
 91        )
 92        matching_sources: list[UserSettingSerializer] = []
 93        for source in _all_sources:
 94            user_settings = source.ui_user_settings()
 95            if not user_settings:
 96                continue
 97            policy_engine = PolicyEngine(source, request.user, request)
 98            policy_engine.build()
 99            if not policy_engine.passing:
100                continue
101            source_settings = source.ui_user_settings()
102            source_settings.initial_data["object_uid"] = source.slug
103            if not source_settings.is_valid():
104                LOGGER.warning(source_settings.errors)
105            matching_sources.append(source_settings.validated_data)
106        return Response(matching_sources)

Get all sources the user can configure

def destroy(self, request: rest_framework.request.Request, *args, **kwargs): 
108    def destroy(self, request: Request, *args, **kwargs):
109        """Prevent deletion of built-in sources"""
110        instance: Source = self.get_object()
111
112        if instance.managed == Source.MANAGED_INBUILT:
113            raise ValidationError(
114                {"detail": "Built-in sources cannot be deleted"}, code="protected"
115            )
116
117        return super().destroy(request, *args, **kwargs)

Prevent deletion of built-in sources

name = None
description = None
suffix = None
detail = None
basename = None
Inherited Members
authentik.core.api.object_types.TypesMixin
types
authentik.core.api.used_by.UsedByMixin
used_by
class UserSourceConnectionSerializer(SourceSerializer): 
120class UserSourceConnectionSerializer(SourceSerializer):
121    """User source connection"""
122
123    source_obj = SourceSerializer(read_only=True, source="source")
124
125    class Meta:
126        model = UserSourceConnection
127        fields = [
128            "pk",
129            "user",
130            "source",
131            "source_obj",
132            "identifier",
133            "created",
134            "last_updated",
135        ]
136        extra_kwargs = {
137            "created": {"read_only": True},
138            "last_updated": {"read_only": True},
139        }

User source connection

source_obj
Inherited Members
SourceSerializer
managed
component
icon_url
icon_themed_urls
get_component
authentik.core.api.utils.ModelSerializer
serializer_field_mapping
update
authentik.core.api.utils.MetaNameSerializer
verbose_name
verbose_name_plural
meta_model_name
get_verbose_name
get_verbose_name_plural
get_meta_model_name
class UserSourceConnectionSerializer.Meta: 
125    class Meta:
126        model = UserSourceConnection
127        fields = [
128            "pk",
129            "user",
130            "source",
131            "source_obj",
132            "identifier",
133            "created",
134            "last_updated",
135        ]
136        extra_kwargs = {
137            "created": {"read_only": True},
138            "last_updated": {"read_only": True},
139        }
fields = ['pk', 'user', 'source', 'source_obj', 'identifier', 'created', 'last_updated']
extra_kwargs = {'created': {'read_only': True}, 'last_updated': {'read_only': True}}
class UserSourceConnectionViewSet(rest_framework.mixins.RetrieveModelMixin, rest_framework.mixins.UpdateModelMixin, rest_framework.mixins.DestroyModelMixin, authentik.core.api.used_by.UsedByMixin, rest_framework.mixins.ListModelMixin, rest_framework.viewsets.GenericViewSet): 
142class UserSourceConnectionViewSet(
143    mixins.RetrieveModelMixin,
144    mixins.UpdateModelMixin,
145    mixins.DestroyModelMixin,
146    UsedByMixin,
147    mixins.ListModelMixin,
148    GenericViewSet,
149):
150    """User-source connection Viewset"""
151
152    queryset = UserSourceConnection.objects.all()
153    serializer_class = UserSourceConnectionSerializer
154    filterset_fields = ["user", "source__slug"]
155    search_fields = ["user__username", "source__slug", "identifier"]
156    ordering = ["source__slug", "pk"]
157    owner_field = "user"

User-source connection Viewset

queryset = <InheritanceQuerySet []>
serializer_class = <class 'UserSourceConnectionSerializer'>
filterset_fields = ['user', 'source__slug']
search_fields = ['user__username', 'source__slug', 'identifier']
ordering = ['source__slug', 'pk']
owner_field = 'user'
name = None
description = None
suffix = None
detail = None
basename = None
Inherited Members
authentik.core.api.used_by.UsedByMixin
used_by
class GroupSourceConnectionSerializer(SourceSerializer): 
160class GroupSourceConnectionSerializer(SourceSerializer):
161    """Group Source Connection"""
162
163    source_obj = SourceSerializer(read_only=True)
164
165    class Meta:
166        model = GroupSourceConnection
167        fields = [
168            "pk",
169            "group",
170            "source",
171            "source_obj",
172            "identifier",
173            "created",
174            "last_updated",
175        ]
176        extra_kwargs = {
177            "created": {"read_only": True},
178            "last_updated": {"read_only": True},
179        }

Group Source Connection

source_obj
Inherited Members
SourceSerializer
managed
component
icon_url
icon_themed_urls
get_component
authentik.core.api.utils.ModelSerializer
serializer_field_mapping
update
authentik.core.api.utils.MetaNameSerializer
verbose_name
verbose_name_plural
meta_model_name
get_verbose_name
get_verbose_name_plural
get_meta_model_name
class GroupSourceConnectionSerializer.Meta: 
165    class Meta:
166        model = GroupSourceConnection
167        fields = [
168            "pk",
169            "group",
170            "source",
171            "source_obj",
172            "identifier",
173            "created",
174            "last_updated",
175        ]
176        extra_kwargs = {
177            "created": {"read_only": True},
178            "last_updated": {"read_only": True},
179        }
fields = ['pk', 'group', 'source', 'source_obj', 'identifier', 'created', 'last_updated']
extra_kwargs = {'created': {'read_only': True}, 'last_updated': {'read_only': True}}
class GroupSourceConnectionViewSet(rest_framework.mixins.RetrieveModelMixin, rest_framework.mixins.UpdateModelMixin, rest_framework.mixins.DestroyModelMixin, authentik.core.api.used_by.UsedByMixin, rest_framework.mixins.ListModelMixin, rest_framework.viewsets.GenericViewSet): 
182class GroupSourceConnectionViewSet(
183    mixins.RetrieveModelMixin,
184    mixins.UpdateModelMixin,
185    mixins.DestroyModelMixin,
186    UsedByMixin,
187    mixins.ListModelMixin,
188    GenericViewSet,
189):
190    """Group-source connection Viewset"""
191
192    queryset = GroupSourceConnection.objects.all()
193    serializer_class = GroupSourceConnectionSerializer
194    filterset_fields = ["group", "source__slug"]
195    search_fields = ["group__name", "source__slug", "identifier"]
196    ordering = ["source__slug", "pk"]

Group-source connection Viewset

queryset = <InheritanceQuerySet []>
serializer_class = <class 'GroupSourceConnectionSerializer'>
filterset_fields = ['group', 'source__slug']
search_fields = ['group__name', 'source__slug', 'identifier']
ordering = ['source__slug', 'pk']
name = None
description = None
suffix = None
detail = None
basename = None
Inherited Members
authentik.core.api.used_by.UsedByMixin
used_by