authentik.sources.telegram.api.source

View Source

 1from django.utils.translation import gettext_lazy as _
 2from drf_spectacular.utils import OpenApiResponse, extend_schema
 3from rest_framework.decorators import action
 4from rest_framework.generics import get_object_or_404
 5from rest_framework.permissions import IsAuthenticated
 6from rest_framework.request import Request
 7from rest_framework.response import Response
 8from rest_framework.viewsets import ModelViewSet
 9
10from authentik.core.api.sources import SourceSerializer
11from authentik.core.api.used_by import UsedByMixin
12from authentik.sources.telegram.api.source_connection import UserTelegramSourceConnectionSerializer
13from authentik.sources.telegram.models import TelegramSource, UserTelegramSourceConnection
14from authentik.sources.telegram.telegram import TelegramAuth
15
16
17class TelegramSourceSerializer(SourceSerializer):
18    class Meta:
19        model = TelegramSource
20        fields = SourceSerializer.Meta.fields + [
21            "bot_username",
22            "bot_token",
23            "request_message_access",
24            "pre_authentication_flow",
25        ]
26        extra_kwargs = {
27            "bot_token": {"write_only": True},
28        }
29
30
31class TelegramAuthSerializer(TelegramAuth):
32
33    def __init__(self, *args, **kwargs):
34        self._bot_token = kwargs.pop("bot_token", None)
35        super().__init__(*args, **kwargs)
36
37    def get_bot_token(self):
38        return self._bot_token
39
40
41class TelegramSourceViewSet(UsedByMixin, ModelViewSet):
42    queryset = TelegramSource.objects.all()
43    serializer_class = TelegramSourceSerializer
44    lookup_field = "slug"
45
46    filterset_fields = [
47        "pbm_uuid",
48        "name",
49        "slug",
50        "enabled",
51        "authentication_flow",
52        "enrollment_flow",
53        "policy_engine_mode",
54        "user_matching_mode",
55        "group_matching_mode",
56        "bot_username",
57        "request_message_access",
58    ]
59    search_fields = ["name", "slug"]
60    ordering = ["name"]
61
62    @extend_schema(
63        request=TelegramAuthSerializer,
64        responses={
65            201: UserTelegramSourceConnectionSerializer,
66            403: OpenApiResponse(description="Access denied"),
67        },
68    )
69    @action(
70        methods=["POST"],
71        url_path="connect_user",
72        detail=True,
73        pagination_class=None,
74        filter_backends=[],
75        permission_classes=[IsAuthenticated],
76    )
77    def connect_user(self, request: Request, slug: str) -> Response:
78
79        source: TelegramSource = get_object_or_404(TelegramSource, slug=slug)
80        serializer = TelegramAuthSerializer(bot_token=source.bot_token, data=request.data)
81        serializer.is_valid(raise_exception=True)
82
83        connection, created = UserTelegramSourceConnection.objects.get_or_create(
84            source=source,
85            identifier=serializer.validated_data["id"],
86            defaults={"user": request.user},
87        )
88        if not created and connection.user != request.user:
89            return Response(
90                data={"detail": _("This Telegram account is already connected to another user.")},
91                status=403,
92            )
93        return Response(
94            data=UserTelegramSourceConnectionSerializer(instance=connection).data, status=201
95        )

class TelegramSourceSerializer(authentik.core.api.sources.SourceSerializer): View Source

18class TelegramSourceSerializer(SourceSerializer):
19    class Meta:
20        model = TelegramSource
21        fields = SourceSerializer.Meta.fields + [
22            "bot_username",
23            "bot_token",
24            "request_message_access",
25            "pre_authentication_flow",
26        ]
27        extra_kwargs = {
28            "bot_token": {"write_only": True},
29        }

Source Serializer

Inherited Members

authentik.core.api.sources.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 TelegramSourceSerializer.Meta: View Source

19    class Meta:
20        model = TelegramSource
21        fields = SourceSerializer.Meta.fields + [
22            "bot_username",
23            "bot_token",
24            "request_message_access",
25            "pre_authentication_flow",
26        ]
27        extra_kwargs = {
28            "bot_token": {"write_only": True},
29        }

model = <class 'authentik.sources.telegram.models.TelegramSource'>

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', 'bot_username', 'bot_token', 'request_message_access', 'pre_authentication_flow']

extra_kwargs = {'bot_token': {'write_only': True}}

class TelegramAuthSerializer(authentik.sources.telegram.telegram.TelegramAuth): View Source

32class TelegramAuthSerializer(TelegramAuth):
33
34    def __init__(self, *args, **kwargs):
35        self._bot_token = kwargs.pop("bot_token", None)
36        super().__init__(*args, **kwargs)
37
38    def get_bot_token(self):
39        return self._bot_token

The BaseSerializer class provides a minimal class which may be used for writing custom serializer implementations.

Note that we strongly restrict the ordering of operations/properties that may be used on the serializer in order to enforce correct usage.

In particular, if a data= argument is passed then:

.is_valid() - Available. .initial_data - Available. .validated_data - Only available after calling is_valid() .errors - Only available after calling is_valid() .data - Only available after calling is_valid()

If a data= argument is not passed then:

.is_valid() - Not available. .initial_data - Not available. .validated_data - Not available. .errors - Not available. .data - Available.

TelegramAuthSerializer(*args, **kwargs) View Source

34    def __init__(self, *args, **kwargs):
35        self._bot_token = kwargs.pop("bot_token", None)
36        super().__init__(*args, **kwargs)

def get_bot_token(self): View Source

38    def get_bot_token(self):
39        return self._bot_token

Inherited Members

authentik.sources.telegram.telegram.TelegramAuth: id; first_name; last_name; username; photo_url; auth_date; hash; validate_auth_date; validate

class TelegramSourceViewSet(authentik.core.api.used_by.UsedByMixin, rest_framework.viewsets.ModelViewSet): View Source

42class TelegramSourceViewSet(UsedByMixin, ModelViewSet):
43    queryset = TelegramSource.objects.all()
44    serializer_class = TelegramSourceSerializer
45    lookup_field = "slug"
46
47    filterset_fields = [
48        "pbm_uuid",
49        "name",
50        "slug",
51        "enabled",
52        "authentication_flow",
53        "enrollment_flow",
54        "policy_engine_mode",
55        "user_matching_mode",
56        "group_matching_mode",
57        "bot_username",
58        "request_message_access",
59    ]
60    search_fields = ["name", "slug"]
61    ordering = ["name"]
62
63    @extend_schema(
64        request=TelegramAuthSerializer,
65        responses={
66            201: UserTelegramSourceConnectionSerializer,
67            403: OpenApiResponse(description="Access denied"),
68        },
69    )
70    @action(
71        methods=["POST"],
72        url_path="connect_user",
73        detail=True,
74        pagination_class=None,
75        filter_backends=[],
76        permission_classes=[IsAuthenticated],
77    )
78    def connect_user(self, request: Request, slug: str) -> Response:
79
80        source: TelegramSource = get_object_or_404(TelegramSource, slug=slug)
81        serializer = TelegramAuthSerializer(bot_token=source.bot_token, data=request.data)
82        serializer.is_valid(raise_exception=True)
83
84        connection, created = UserTelegramSourceConnection.objects.get_or_create(
85            source=source,
86            identifier=serializer.validated_data["id"],
87            defaults={"user": request.user},
88        )
89        if not created and connection.user != request.user:
90            return Response(
91                data={"detail": _("This Telegram account is already connected to another user.")},
92                status=403,
93            )
94        return Response(
95            data=UserTelegramSourceConnectionSerializer(instance=connection).data, status=201
96        )

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

queryset = <InheritanceQuerySet []>

serializer_class = <class 'TelegramSourceSerializer'>

lookup_field = 'slug'

filterset_fields = ['pbm_uuid', 'name', 'slug', 'enabled', 'authentication_flow', 'enrollment_flow', 'policy_engine_mode', 'user_matching_mode', 'group_matching_mode', 'bot_username', 'request_message_access']

search_fields = ['name', 'slug']

ordering = ['name']

@extend_schema(request=TelegramAuthSerializer, responses={201: UserTelegramSourceConnectionSerializer, 403: OpenApiResponse(description='Access denied')})

@action(methods=['POST'], url_path='connect_user', detail=True, pagination_class=None, filter_backends=[], permission_classes=[IsAuthenticated])

def connect_user( self, request: rest_framework.request.Request, slug: str) -> rest_framework.response.Response: View Source

63    @extend_schema(
64        request=TelegramAuthSerializer,
65        responses={
66            201: UserTelegramSourceConnectionSerializer,
67            403: OpenApiResponse(description="Access denied"),
68        },
69    )
70    @action(
71        methods=["POST"],
72        url_path="connect_user",
73        detail=True,
74        pagination_class=None,
75        filter_backends=[],
76        permission_classes=[IsAuthenticated],
77    )
78    def connect_user(self, request: Request, slug: str) -> Response:
79
80        source: TelegramSource = get_object_or_404(TelegramSource, slug=slug)
81        serializer = TelegramAuthSerializer(bot_token=source.bot_token, data=request.data)
82        serializer.is_valid(raise_exception=True)
83
84        connection, created = UserTelegramSourceConnection.objects.get_or_create(
85            source=source,
86            identifier=serializer.validated_data["id"],
87            defaults={"user": request.user},
88        )
89        if not created and connection.user != request.user:
90            return Response(
91                data={"detail": _("This Telegram account is already connected to another user.")},
92                status=403,
93            )
94        return Response(
95            data=UserTelegramSourceConnectionSerializer(instance=connection).data, status=201
96        )

name = None

description = None

suffix = None

detail = None

basename = None

Inherited Members

authentik.core.api.used_by.UsedByMixin: used_by