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            "user": {"read_only": True},
137            "created": {"read_only": True},
138            "last_updated": {"read_only": True},
139        }
140
141
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"
158
159
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            "group": {"read_only": True},
178            "created": {"read_only": True},
179            "last_updated": {"read_only": True},
180        }
181
182
183class GroupSourceConnectionViewSet(
184    mixins.RetrieveModelMixin,
185    mixins.UpdateModelMixin,
186    mixins.DestroyModelMixin,
187    UsedByMixin,
188    mixins.ListModelMixin,
189    GenericViewSet,
190):
191    """Group-source connection Viewset"""
192
193    queryset = GroupSourceConnection.objects.all()
194    serializer_class = GroupSourceConnectionSerializer
195    filterset_fields = ["group", "source__slug"]
196    search_fields = ["group__name", "source__slug", "identifier"]
197    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            "user": {"read_only": True},
138            "created": {"read_only": True},
139            "last_updated": {"read_only": True},
140        }

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            "user": {"read_only": True},
138            "created": {"read_only": True},
139            "last_updated": {"read_only": True},
140        }
fields = ['pk', 'user', 'source', 'source_obj', 'identifier', 'created', 'last_updated']
extra_kwargs = {'user': {'read_only': True}, '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): 
143class UserSourceConnectionViewSet(
144    mixins.RetrieveModelMixin,
145    mixins.UpdateModelMixin,
146    mixins.DestroyModelMixin,
147    UsedByMixin,
148    mixins.ListModelMixin,
149    GenericViewSet,
150):
151    """User-source connection Viewset"""
152
153    queryset = UserSourceConnection.objects.all()
154    serializer_class = UserSourceConnectionSerializer
155    filterset_fields = ["user", "source__slug"]
156    search_fields = ["user__username", "source__slug", "identifier"]
157    ordering = ["source__slug", "pk"]
158    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): 
161class GroupSourceConnectionSerializer(SourceSerializer):
162    """Group Source Connection"""
163
164    source_obj = SourceSerializer(read_only=True)
165
166    class Meta:
167        model = GroupSourceConnection
168        fields = [
169            "pk",
170            "group",
171            "source",
172            "source_obj",
173            "identifier",
174            "created",
175            "last_updated",
176        ]
177        extra_kwargs = {
178            "group": {"read_only": True},
179            "created": {"read_only": True},
180            "last_updated": {"read_only": True},
181        }

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: 
166    class Meta:
167        model = GroupSourceConnection
168        fields = [
169            "pk",
170            "group",
171            "source",
172            "source_obj",
173            "identifier",
174            "created",
175            "last_updated",
176        ]
177        extra_kwargs = {
178            "group": {"read_only": True},
179            "created": {"read_only": True},
180            "last_updated": {"read_only": True},
181        }
fields = ['pk', 'group', 'source', 'source_obj', 'identifier', 'created', 'last_updated']
extra_kwargs = {'group': {'read_only': True}, '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): 
184class GroupSourceConnectionViewSet(
185    mixins.RetrieveModelMixin,
186    mixins.UpdateModelMixin,
187    mixins.DestroyModelMixin,
188    UsedByMixin,
189    mixins.ListModelMixin,
190    GenericViewSet,
191):
192    """Group-source connection Viewset"""
193
194    queryset = GroupSourceConnection.objects.all()
195    serializer_class = GroupSourceConnectionSerializer
196    filterset_fields = ["group", "source__slug"]
197    search_fields = ["group__name", "source__slug", "identifier"]
198    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