authentik.sources.plex.models

  1"""Plex source"""
  2
  3from typing import Any
  4
  5from django.contrib.postgres.fields import ArrayField
  6from django.db import models
  7from django.http.request import HttpRequest
  8from django.templatetags.static import static
  9from django.utils.translation import gettext_lazy as _
 10from rest_framework.fields import CharField
 11from rest_framework.serializers import BaseSerializer, Serializer
 12
 13from authentik.core.models import (
 14    GroupSourceConnection,
 15    PropertyMapping,
 16    Source,
 17    UserSourceConnection,
 18)
 19from authentik.core.types import UILoginButton, UserSettingSerializer
 20from authentik.flows.challenge import Challenge, ChallengeResponse
 21from authentik.lib.generators import generate_id
 22from authentik.lib.utils.time import fqdn_rand
 23from authentik.stages.identification.stage import LoginChallengeMixin
 24from authentik.tasks.schedules.common import ScheduleSpec
 25from authentik.tasks.schedules.models import ScheduledModel
 26
 27
 28class PlexAuthenticationChallenge(LoginChallengeMixin, Challenge):
 29    """Challenge shown to the user in identification stage"""
 30
 31    client_id = CharField()
 32    slug = CharField()
 33    component = CharField(default="ak-source-plex")
 34
 35
 36class PlexAuthenticationChallengeResponse(ChallengeResponse):
 37    """Pseudo class for plex response"""
 38
 39    component = CharField(default="ak-source-plex")
 40
 41
 42class PlexSource(ScheduledModel, Source):
 43    """Authenticate against plex.tv"""
 44
 45    client_id = models.TextField(
 46        default=generate_id,
 47        help_text=_("Client identifier used to talk to Plex."),
 48    )
 49    allowed_servers = ArrayField(
 50        models.TextField(),
 51        default=list,
 52        blank=True,
 53        help_text=_(
 54            "Which servers a user has to be a member of to be granted access. "
 55            "Empty list allows every server."
 56        ),
 57    )
 58    allow_friends = models.BooleanField(
 59        default=True,
 60        help_text=_("Allow friends to authenticate, even if you don't share a server."),
 61    )
 62    plex_token = models.TextField(help_text=_("Plex token used to check friends"))
 63
 64    @property
 65    def component(self) -> str:
 66        return "ak-source-plex-form"
 67
 68    @property
 69    def serializer(self) -> type[BaseSerializer]:
 70        from authentik.sources.plex.api.source import PlexSourceSerializer
 71
 72        return PlexSourceSerializer
 73
 74    @property
 75    def property_mapping_type(self) -> type[PropertyMapping]:
 76        return PlexSourcePropertyMapping
 77
 78    @property
 79    def schedule_specs(self) -> list[ScheduleSpec]:
 80        from authentik.sources.plex.tasks import check_plex_token
 81
 82        return [
 83            ScheduleSpec(
 84                actor=check_plex_token,
 85                uid=self.slug,
 86                args=(self.pk,),
 87                crontab=f"{fqdn_rand(self.pk)} */3 * * *",
 88            ),
 89        ]
 90
 91    def get_base_user_properties(self, info: dict[str, Any], **kwargs):
 92        return {
 93            "username": info.get("username"),
 94            "email": info.get("email"),
 95            "name": info.get("title"),
 96        }
 97
 98    def get_base_group_properties(self, group_id: str, **kwargs):
 99        return {
100            "name": group_id,
101        }
102
103    @property
104    def icon_url(self) -> str:
105        icon = super().icon_url
106        if not icon:
107            icon = static("authentik/sources/plex.svg")
108        return icon
109
110    def ui_login_button(self, request: HttpRequest) -> UILoginButton:
111        return UILoginButton(
112            challenge=PlexAuthenticationChallenge(
113                data={
114                    "component": "ak-source-plex",
115                    "client_id": self.client_id,
116                    "slug": self.slug,
117                }
118            ),
119            icon_url=self.icon_url,
120            name=self.name,
121            promoted=self.promoted,
122        )
123
124    def ui_user_settings(self) -> UserSettingSerializer | None:
125        return UserSettingSerializer(
126            data={
127                "title": self.name,
128                "component": "ak-user-settings-source-plex",
129                "configure_url": self.client_id,
130                "icon_url": self.icon_url,
131            }
132        )
133
134    class Meta:
135        verbose_name = _("Plex Source")
136        verbose_name_plural = _("Plex Sources")
137
138
139class PlexSourcePropertyMapping(PropertyMapping):
140    """Map Plex properties to User of Group object attributes"""
141
142    @property
143    def component(self) -> str:
144        return "ak-property-mapping-source-plex-form"
145
146    @property
147    def serializer(self) -> type[Serializer]:
148        from authentik.sources.plex.api.property_mappings import PlexSourcePropertyMappingSerializer
149
150        return PlexSourcePropertyMappingSerializer
151
152    class Meta:
153        verbose_name = _("Plex Source Property Mapping")
154        verbose_name_plural = _("Plex Source Property Mappings")
155
156
157class UserPlexSourceConnection(UserSourceConnection):
158    """Connect user and plex source"""
159
160    plex_token = models.TextField()
161
162    @property
163    def serializer(self) -> type[Serializer]:
164        from authentik.sources.plex.api.source_connection import UserPlexSourceConnectionSerializer
165
166        return UserPlexSourceConnectionSerializer
167
168    class Meta:
169        verbose_name = _("User Plex Source Connection")
170        verbose_name_plural = _("User Plex Source Connections")
171
172
173class GroupPlexSourceConnection(GroupSourceConnection):
174    """Group-source connection"""
175
176    @property
177    def serializer(self) -> type[Serializer]:
178        from authentik.sources.plex.api.source_connection import (
179            GroupPlexSourceConnectionSerializer,
180        )
181
182        return GroupPlexSourceConnectionSerializer
183
184    class Meta:
185        verbose_name = _("Group Plex Source Connection")
186        verbose_name_plural = _("Group Plex Source Connections")