authentik.outposts.api.service_connections

  1"""Outpost API Views"""
  2
  3from dataclasses import asdict
  4
  5from django.utils.translation import gettext_lazy as _
  6from drf_spectacular.utils import extend_schema
  7from kubernetes.client.configuration import Configuration
  8from kubernetes.config.config_exception import ConfigException
  9from kubernetes.config.kube_config import load_kube_config_from_dict
 10from rest_framework import mixins, serializers
 11from rest_framework.decorators import action
 12from rest_framework.fields import BooleanField, CharField, ReadOnlyField
 13from rest_framework.request import Request
 14from rest_framework.response import Response
 15from rest_framework.viewsets import GenericViewSet, ModelViewSet
 16
 17from authentik.core.api.object_types import TypesMixin
 18from authentik.core.api.used_by import UsedByMixin
 19from authentik.core.api.utils import (
 20    MetaNameSerializer,
 21    ModelSerializer,
 22    PassiveSerializer,
 23)
 24from authentik.outposts.models import (
 25    DockerServiceConnection,
 26    KubernetesServiceConnection,
 27    OutpostServiceConnection,
 28)
 29from authentik.rbac.filters import ObjectFilter
 30
 31
 32class ServiceConnectionSerializer(ModelSerializer, MetaNameSerializer):
 33    """ServiceConnection Serializer"""
 34
 35    component = ReadOnlyField()
 36
 37    def get_component(self, obj: OutpostServiceConnection) -> str:
 38        """Get object type so that we know how to edit the object"""
 39        if obj.__class__ == OutpostServiceConnection:
 40            return ""
 41        return obj.component
 42
 43    class Meta:
 44        model = OutpostServiceConnection
 45        fields = [
 46            "pk",
 47            "name",
 48            "local",
 49            "component",
 50            "verbose_name",
 51            "verbose_name_plural",
 52            "meta_model_name",
 53        ]
 54
 55
 56class ServiceConnectionStateSerializer(PassiveSerializer):
 57    """Serializer for Service connection state"""
 58
 59    healthy = BooleanField(read_only=True)
 60    version = CharField(read_only=True)
 61
 62
 63class ServiceConnectionViewSet(
 64    TypesMixin,
 65    mixins.RetrieveModelMixin,
 66    mixins.DestroyModelMixin,
 67    UsedByMixin,
 68    mixins.ListModelMixin,
 69    GenericViewSet,
 70):
 71    """ServiceConnection Viewset"""
 72
 73    queryset = OutpostServiceConnection.objects.select_subclasses()
 74    serializer_class = ServiceConnectionSerializer
 75    search_fields = ["name"]
 76    filterset_fields = ["name"]
 77
 78    @extend_schema(responses={200: ServiceConnectionStateSerializer(many=False)})
 79    @action(detail=True, pagination_class=None, filter_backends=[ObjectFilter])
 80    def state(self, request: Request, pk: str) -> Response:
 81        """Get the service connection's state"""
 82        connection = self.get_object()
 83        return Response(asdict(connection.state))
 84
 85
 86class DockerServiceConnectionSerializer(ServiceConnectionSerializer):
 87    """DockerServiceConnection Serializer"""
 88
 89    class Meta:
 90        model = DockerServiceConnection
 91        fields = ServiceConnectionSerializer.Meta.fields + [
 92            "url",
 93            "tls_verification",
 94            "tls_authentication",
 95        ]
 96
 97
 98class DockerServiceConnectionViewSet(UsedByMixin, ModelViewSet):
 99    """DockerServiceConnection Viewset"""
100
101    queryset = DockerServiceConnection.objects.all()
102    serializer_class = DockerServiceConnectionSerializer
103    filterset_fields = ["name", "local", "url", "tls_verification", "tls_authentication"]
104    ordering = ["name"]
105    search_fields = ["name"]
106
107
108class KubernetesServiceConnectionSerializer(ServiceConnectionSerializer):
109    """KubernetesServiceConnection Serializer"""
110
111    def validate_kubeconfig(self, kubeconfig):
112        """Validate kubeconfig by attempting to load it"""
113        if kubeconfig == {}:
114            if not self.initial_data["local"]:
115                raise serializers.ValidationError(
116                    _("You can only use an empty kubeconfig when connecting to a local cluster.")
117                )
118            # Empty kubeconfig is valid
119            return kubeconfig
120        config = Configuration()
121        try:
122            load_kube_config_from_dict(kubeconfig, client_configuration=config)
123        except ConfigException:
124            raise serializers.ValidationError(_("Invalid kubeconfig")) from None
125        return kubeconfig
126
127    class Meta:
128        model = KubernetesServiceConnection
129        fields = ServiceConnectionSerializer.Meta.fields + ["kubeconfig", "verify_ssl"]
130
131
132class KubernetesServiceConnectionViewSet(UsedByMixin, ModelViewSet):
133    """KubernetesServiceConnection Viewset"""
134
135    queryset = KubernetesServiceConnection.objects.all()
136    serializer_class = KubernetesServiceConnectionSerializer
137    filterset_fields = ["name", "local"]
138    ordering = ["name"]
139    search_fields = ["name"]