authentik.outposts.api.outposts

  1"""Outpost API Views"""
  2
  3from dacite.core import from_dict
  4from dacite.exceptions import DaciteError
  5from django_filters.filters import ModelMultipleChoiceFilter
  6from django_filters.filterset import FilterSet
  7from drf_spectacular.utils import extend_schema
  8from rest_framework.decorators import action
  9from rest_framework.exceptions import ValidationError
 10from rest_framework.fields import BooleanField, CharField, DateTimeField, SerializerMethodField
 11from rest_framework.relations import PrimaryKeyRelatedField
 12from rest_framework.request import Request
 13from rest_framework.response import Response
 14from rest_framework.viewsets import ModelViewSet
 15
 16from authentik import authentik_build_hash
 17from authentik.admin.api.system import fips_enabled
 18from authentik.core.api.providers import ProviderSerializer
 19from authentik.core.api.used_by import UsedByMixin
 20from authentik.core.api.utils import JSONDictField, ModelSerializer, PassiveSerializer
 21from authentik.core.models import Provider
 22from authentik.lib.utils.time import timedelta_from_string, timedelta_string_validator
 23from authentik.outposts.api.service_connections import ServiceConnectionSerializer
 24from authentik.outposts.apps import MANAGED_OUTPOST, MANAGED_OUTPOST_NAME
 25from authentik.outposts.models import (
 26    Outpost,
 27    OutpostConfig,
 28    OutpostType,
 29    default_outpost_config,
 30)
 31from authentik.providers.ldap.models import LDAPProvider
 32from authentik.providers.proxy.models import ProxyProvider
 33from authentik.providers.rac.models import RACProvider
 34from authentik.providers.radius.models import RadiusProvider
 35
 36
 37class OutpostSerializer(ModelSerializer):
 38    """Outpost Serializer"""
 39
 40    config = JSONDictField(source="_config")
 41    # Need to set allow_empty=True for the embedded outpost with no providers
 42    # is checked for other providers in the API Viewset
 43    providers = PrimaryKeyRelatedField(
 44        allow_empty=True,
 45        many=True,
 46        queryset=Provider.objects.select_subclasses().all(),
 47    )
 48    providers_obj = ProviderSerializer(source="providers", many=True, read_only=True)
 49    service_connection_obj = ServiceConnectionSerializer(
 50        source="service_connection",
 51        read_only=True,
 52        allow_null=True,
 53    )
 54    refresh_interval_s = SerializerMethodField()
 55
 56    def get_refresh_interval_s(self, obj: Outpost) -> int:
 57        return int(timedelta_from_string(obj.config.refresh_interval).total_seconds())
 58
 59    def validate_name(self, name: str) -> str:
 60        """Validate name (especially for embedded outpost)"""
 61        if not self.instance:
 62            return name
 63        if self.instance.managed == MANAGED_OUTPOST and name != MANAGED_OUTPOST_NAME:
 64            raise ValidationError("Embedded outpost's name cannot be changed")
 65        if self.instance.name == MANAGED_OUTPOST_NAME:
 66            self.instance.managed = MANAGED_OUTPOST
 67        return name
 68
 69    def validate_providers(self, providers: list[Provider]) -> list[Provider]:
 70        """Check that all providers match the type of the outpost"""
 71        type_map = {
 72            OutpostType.LDAP: LDAPProvider,
 73            OutpostType.PROXY: ProxyProvider,
 74            OutpostType.RADIUS: RadiusProvider,
 75            OutpostType.RAC: RACProvider,
 76            None: Provider,
 77        }
 78        for provider in providers:
 79            if not isinstance(provider, type_map[self.initial_data.get("type")]):
 80                raise ValidationError(
 81                    f"Outpost type {self.initial_data['type']} can't be used with "
 82                    f"{provider.__class__.__name__} providers."
 83                )
 84        if self.instance and self.instance.managed == MANAGED_OUTPOST:
 85            return providers
 86        if len(providers) < 1:
 87            raise ValidationError("This list may not be empty.")
 88        return providers
 89
 90    def validate_config(self, config) -> dict:
 91        """Check that the config has all required fields"""
 92        try:
 93            parsed = from_dict(OutpostConfig, config)
 94            timedelta_string_validator(parsed.refresh_interval)
 95        except DaciteError as exc:
 96            raise ValidationError(f"Failed to validate config: {str(exc)}") from exc
 97        return config
 98
 99    class Meta:
100        model = Outpost
101        fields = [
102            "pk",
103            "name",
104            "type",
105            "providers",
106            "providers_obj",
107            "service_connection",
108            "service_connection_obj",
109            "refresh_interval_s",
110            "token_identifier",
111            "config",
112            "managed",
113        ]
114        extra_kwargs = {"type": {"required": True}}
115
116
117class OutpostDefaultConfigSerializer(PassiveSerializer):
118    """Global default outpost config"""
119
120    config = JSONDictField(read_only=True)
121
122
123class OutpostHealthSerializer(PassiveSerializer):
124    """Outpost health status"""
125
126    uid = CharField(read_only=True)
127    last_seen = DateTimeField(read_only=True)
128    version = CharField(read_only=True)
129    golang_version = CharField(read_only=True)
130    openssl_enabled = BooleanField(read_only=True)
131    openssl_version = CharField(read_only=True)
132    fips_enabled = SerializerMethodField()
133
134    version_should = CharField(read_only=True)
135    version_outdated = BooleanField(read_only=True)
136
137    build_hash = CharField(read_only=True, required=False)
138    build_hash_should = CharField(read_only=True, required=False)
139
140    hostname = CharField(read_only=True, required=False)
141
142    def get_fips_enabled(self, obj: dict) -> bool | None:
143        """Get FIPS enabled"""
144        if not fips_enabled():
145            return None
146        return obj["fips_enabled"]
147
148
149class OutpostFilter(FilterSet):
150    """Filter for Outposts"""
151
152    providers_by_pk = ModelMultipleChoiceFilter(
153        field_name="providers",
154        queryset=Provider.objects.all(),
155    )
156
157    class Meta:
158        model = Outpost
159        fields = {
160            "providers": ["isnull"],
161            "name": ["iexact", "icontains"],
162            "service_connection__name": ["iexact", "icontains"],
163            "managed": ["iexact", "icontains"],
164        }
165
166
167class OutpostViewSet(UsedByMixin, ModelViewSet):
168    """Outpost Viewset"""
169
170    queryset = Outpost.objects.all()
171    serializer_class = OutpostSerializer
172    filterset_class = OutpostFilter
173    search_fields = [
174        "name",
175        "providers__name",
176    ]
177    ordering = ["name", "service_connection__name"]
178
179    @extend_schema(responses={200: OutpostHealthSerializer(many=True)})
180    @action(methods=["GET"], detail=True, pagination_class=None)
181    def health(self, request: Request, pk: int) -> Response:
182        """Get outposts current health"""
183        outpost: Outpost = self.get_object()
184        states = []
185        for state in outpost.state:
186            states.append(
187                {
188                    "uid": state.uid,
189                    "last_seen": state.last_seen,
190                    "version": state.version,
191                    "version_should": state.version_should,
192                    "version_outdated": state.version_outdated,
193                    "build_hash": state.build_hash,
194                    "golang_version": state.golang_version,
195                    "openssl_enabled": state.openssl_enabled,
196                    "openssl_version": state.openssl_version,
197                    "fips_enabled": state.fips_enabled,
198                    "hostname": state.hostname,
199                    "build_hash_should": authentik_build_hash(),
200                }
201            )
202        return Response(OutpostHealthSerializer(states, many=True).data)
203
204    @extend_schema(responses={200: OutpostDefaultConfigSerializer(many=False)})
205    @action(detail=False, methods=["GET"])
206    def default_settings(self, request: Request) -> Response:
207        """Global default outpost config"""
208        host = self.request.build_absolute_uri("/")
209        return Response({"config": default_outpost_config(host)})