authentik.providers.rac.models

  1"""RAC Models"""
  2
  3from typing import Any
  4from uuid import uuid4
  5
  6from deepmerge import always_merger
  7from django.db import models
  8from django.db.models import QuerySet
  9from django.http import HttpRequest
 10from django.templatetags.static import static
 11from django.utils.translation import gettext as _
 12from rest_framework.serializers import Serializer
 13from structlog.stdlib import get_logger
 14
 15from authentik.core.expression.exceptions import PropertyMappingExpressionException
 16from authentik.core.models import ExpiringModel, PropertyMapping, Provider, User, default_token_key
 17from authentik.events.models import Event, EventAction
 18from authentik.lib.models import InternallyManagedMixin, SerializerModel
 19from authentik.lib.utils.time import timedelta_string_validator
 20from authentik.outposts.models import OutpostModel
 21from authentik.policies.models import PolicyBindingModel
 22
 23LOGGER = get_logger()
 24
 25
 26class Protocols(models.TextChoices):
 27    """Supported protocols"""
 28
 29    RDP = "rdp"
 30    VNC = "vnc"
 31    SSH = "ssh"
 32
 33
 34class AuthenticationMode(models.TextChoices):
 35    """Authentication modes"""
 36
 37    STATIC = "static"
 38    PROMPT = "prompt"
 39
 40
 41class RACProvider(OutpostModel, Provider):
 42    """Remotely access computers/servers via RDP/SSH/VNC."""
 43
 44    settings = models.JSONField(default=dict)
 45    auth_mode = models.TextField(
 46        choices=AuthenticationMode.choices, default=AuthenticationMode.PROMPT
 47    )
 48    connection_expiry = models.TextField(
 49        default="hours=8",
 50        validators=[timedelta_string_validator],
 51        help_text=_(
 52            "Determines how long a session lasts. Default of 0 means "
 53            "that the sessions lasts until the browser is closed. "
 54            "(Format: hours=-1;minutes=-2;seconds=-3)"
 55        ),
 56    )
 57    delete_token_on_disconnect = models.BooleanField(
 58        default=False,
 59        help_text=_("When set to true, connection tokens will be deleted upon disconnect."),
 60    )
 61
 62    @property
 63    def launch_url(self) -> str | None:
 64        """URL to this provider and initiate authorization for the user.
 65        Can return None for providers that are not URL-based"""
 66        return "goauthentik.io://providers/rac/launch"
 67
 68    @property
 69    def icon_url(self) -> str | None:
 70        return static("authentik/sources/rac.svg")
 71
 72    @property
 73    def component(self) -> str:
 74        return "ak-provider-rac-form"
 75
 76    @property
 77    def serializer(self) -> type[Serializer]:
 78        from authentik.providers.rac.api.providers import RACProviderSerializer
 79
 80        return RACProviderSerializer
 81
 82    class Meta:
 83        verbose_name = _("RAC Provider")
 84        verbose_name_plural = _("RAC Providers")
 85
 86
 87class Endpoint(SerializerModel, PolicyBindingModel):
 88    """Remote-accessible endpoint"""
 89
 90    name = models.TextField()
 91    host = models.TextField()
 92    protocol = models.TextField(choices=Protocols.choices)
 93    settings = models.JSONField(default=dict)
 94    auth_mode = models.TextField(choices=AuthenticationMode.choices)
 95    provider = models.ForeignKey("RACProvider", on_delete=models.CASCADE)
 96    maximum_connections = models.IntegerField(default=1)
 97
 98    property_mappings = models.ManyToManyField(
 99        "authentik_core.PropertyMapping", default=None, blank=True
100    )
101
102    @property
103    def serializer(self) -> type[Serializer]:
104        from authentik.providers.rac.api.endpoints import EndpointSerializer
105
106        return EndpointSerializer
107
108    def __str__(self):
109        return f"RAC Endpoint {self.name}"
110
111    class Meta:
112        verbose_name = _("RAC Endpoint")
113        verbose_name_plural = _("RAC Endpoints")
114
115
116class RACPropertyMapping(PropertyMapping):
117    """Configure settings for remote access endpoints."""
118
119    static_settings = models.JSONField(default=dict)
120
121    def evaluate(self, user: User | None, request: HttpRequest | None, **kwargs) -> Any:
122        """Evaluate `self.expression` using `**kwargs` as Context."""
123        settings = {}
124        for key, value in self.static_settings.items():
125            if value and value != "":
126                settings[key] = value
127        if self.expression != "":
128            always_merger.merge(settings, super().evaluate(user, request, **kwargs))
129        return settings
130
131    @property
132    def component(self) -> str:
133        return "ak-property-mapping-provider-rac-form"
134
135    @property
136    def serializer(self) -> type[Serializer]:
137        from authentik.providers.rac.api.property_mappings import (
138            RACPropertyMappingSerializer,
139        )
140
141        return RACPropertyMappingSerializer
142
143    class Meta:
144        verbose_name = _("RAC Provider Property Mapping")
145        verbose_name_plural = _("RAC Provider Property Mappings")
146
147
148class ConnectionToken(InternallyManagedMixin, ExpiringModel):
149    """Token for a single connection to a specified endpoint"""
150
151    connection_token_uuid = models.UUIDField(default=uuid4, primary_key=True)
152    provider = models.ForeignKey(RACProvider, on_delete=models.CASCADE)
153    endpoint = models.ForeignKey(Endpoint, on_delete=models.CASCADE)
154    token = models.TextField(default=default_token_key)
155    settings = models.JSONField(default=dict)
156    session = models.ForeignKey("authentik_core.AuthenticatedSession", on_delete=models.CASCADE)
157
158    def get_settings(self) -> dict:
159        """Get settings"""
160        default_settings = {}
161        if ":" in self.endpoint.host:
162            host, _, port = self.endpoint.host.partition(":")
163            default_settings["hostname"] = host
164            default_settings["port"] = str(port)
165        else:
166            default_settings["hostname"] = self.endpoint.host
167        if self.endpoint.protocol == Protocols.RDP:
168            default_settings["resize-method"] = "display-update"
169        default_settings["client-name"] = f"authentik - {self.session.user}"
170        settings = {}
171        always_merger.merge(settings, default_settings)
172        always_merger.merge(settings, self.endpoint.provider.settings)
173        always_merger.merge(settings, self.endpoint.settings)
174
175        def mapping_evaluator(mappings: QuerySet):
176            for mapping in mappings:
177                mapping: RACPropertyMapping
178                try:
179                    mapping_settings = mapping.evaluate(
180                        self.session.user, None, endpoint=self.endpoint, provider=self.provider
181                    )
182                    always_merger.merge(settings, mapping_settings)
183                except PropertyMappingExpressionException as exc:
184                    Event.new(
185                        EventAction.CONFIGURATION_ERROR,
186                        message=f"Failed to evaluate property-mapping: '{mapping.name}'",
187                        provider=self.provider,
188                        mapping=mapping,
189                    ).set_user(self.session.user).save()
190                    LOGGER.warning("Failed to evaluate property mapping", exc=exc)
191
192        mapping_evaluator(
193            RACPropertyMapping.objects.filter(provider__in=[self.provider]).order_by("name")
194        )
195        mapping_evaluator(
196            RACPropertyMapping.objects.filter(endpoint__in=[self.endpoint]).order_by("name")
197        )
198        always_merger.merge(settings, self.settings)
199
200        settings["drive-path"] = f"/tmp/connection/{self.token}"  # nosec
201        settings["create-drive-path"] = "true"
202        # Ensure all values of the settings dict are strings
203        for key, value in settings.items():
204            if isinstance(value, str):
205                continue
206            # Special case for bools
207            if isinstance(value, bool):
208                settings[key] = str(value).lower()
209                continue
210            settings[key] = str(value)
211        return settings
212
213    def __str__(self):
214        return f"RAC Connection token {self.session_id} to {self.provider_id}/{self.endpoint_id}"
215
216    class Meta:
217        verbose_name = _("RAC Connection token")
218        verbose_name_plural = _("RAC Connection tokens")
219        indexes = ExpiringModel.Meta.indexes