authentik/authentik/brands/api.py

"""Serializer for brands models"""

from typing import Any, get_args

from django.db import models
from drf_spectacular.extensions import OpenApiSerializerFieldExtension
from drf_spectacular.plumbing import build_basic_type, build_object_type
from drf_spectacular.utils import extend_schema, extend_schema_field
from rest_framework.decorators import action
from rest_framework.exceptions import ValidationError
from rest_framework.fields import (
    CharField,
    ChoiceField,
    ListField,
    SerializerMethodField,
)
from rest_framework.filters import OrderingFilter, SearchFilter
from rest_framework.permissions import AllowAny
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.validators import UniqueValidator
from rest_framework.viewsets import ModelViewSet

from authentik.brands.models import Brand
from authentik.brands.utils import session_safe_mode
from authentik.core.api.used_by import UsedByMixin
from authentik.core.api.utils import ModelSerializer, PassiveSerializer, ThemedUrlsSerializer
from authentik.rbac.filters import SecretKeyFilter
from authentik.tenants.api.settings import FlagJSONField
from authentik.tenants.flags import Flag
from authentik.tenants.utils import get_current_tenant


class FooterLinkSerializer(PassiveSerializer):
    """Links returned in Config API"""

    href = CharField(read_only=True, allow_null=True)
    name = CharField(read_only=True)


class BrandSerializer(ModelSerializer):
    """Brand Serializer"""

    def validate(self, attrs: dict[str, Any]) -> dict[str, Any]:
        if attrs.get("default", False):
            brands = Brand.objects.filter(default=True)
            if self.instance:
                brands = brands.exclude(pk=self.instance.pk)
            if brands.exists():
                raise ValidationError({"default": "Only a single brand can be set as default."})
        return super().validate(attrs)

    class Meta:
        model = Brand
        fields = [
            "brand_uuid",
            "domain",
            "default",
            "branding_title",
            "branding_logo",
            "branding_favicon",
            "branding_custom_css",
            "branding_default_flow_background",
            "branding_map_tiles",
            "flow_authentication",
            "flow_user_switch",
            "flow_invalidation",
            "flow_recovery",
            "flow_unenrollment",
            "flow_user_settings",
            "flow_device_code",
            "flow_lockdown",
            "flow_request",
            "default_application",
            "web_certificate",
            "client_certificates",
            "attributes",
        ]
        extra_kwargs = {
            # TODO: This field isn't unique on the database which is hard to backport
            # hence we just validate the uniqueness here
            "domain": {"validators": [UniqueValidator(Brand.objects.all())]},
        }


class Themes(models.TextChoices):
    """Themes"""

    AUTOMATIC = "automatic"
    LIGHT = "light"
    DARK = "dark"


def get_default_ui_footer_links():
    """Get default UI footer links based on current tenant settings"""
    return get_current_tenant().footer_links


class PublicFlagsField(FlagJSONField):
    pass


class FlagsJSONExtension(OpenApiSerializerFieldExtension):
    """Generate API Schema for JSON fields as"""

    target_class = "authentik.brands.api.PublicFlagsField"

    def map_serializer_field(self, auto_schema, direction):
        props = {}
        # Public flags are always present; authenticated flags are only present for
        # authenticated requests, so they are exposed in the schema but not required.
        required = []
        for visibility in ("public", "authenticated"):
            for flag in Flag.available(visibility=visibility):
                _flag = flag()
                props[_flag.key] = build_basic_type(get_args(_flag.__orig_bases__[0])[0])
                if _flag.description:
                    props[_flag.key]["description"] = _flag.description
                if _flag.deprecated:
                    props[_flag.key]["deprecated"] = _flag.deprecated
                if visibility == "public" and not _flag.deprecated:
                    required.append(_flag.key)
        return build_object_type(props, required=required)


class CurrentBrandSerializer(PassiveSerializer):
    """Partial brand information for styling"""

    matched_domain = CharField(source="domain")
    branding_title = CharField()
    branding_logo = CharField(source="branding_logo_url")
    branding_logo_themed_urls = ThemedUrlsSerializer(read_only=True, allow_null=True)
    branding_favicon = CharField(source="branding_favicon_url")
    branding_favicon_themed_urls = ThemedUrlsSerializer(read_only=True, allow_null=True)
    branding_custom_css = CharField()
    branding_map_tiles = CharField()
    ui_footer_links = ListField(
        child=FooterLinkSerializer(),
        read_only=True,
        default=get_default_ui_footer_links,
    )
    ui_theme = ChoiceField(
        choices=Themes.choices,
        source="attributes.settings.theme.base",
        default=Themes.AUTOMATIC,
        read_only=True,
    )

    flow_authentication = CharField(source="flow_authentication.slug", required=False)
    flow_user_switch = CharField(source="flow_user_switch.slug", required=False)
    flow_invalidation = CharField(source="flow_invalidation.slug", required=False)
    flow_recovery = CharField(source="flow_recovery.slug", required=False)
    flow_unenrollment = CharField(source="flow_unenrollment.slug", required=False)
    flow_user_settings = CharField(source="flow_user_settings.slug", required=False)
    flow_device_code = CharField(source="flow_device_code.slug", required=False)
    flow_lockdown = CharField(source="flow_lockdown.slug", required=False)
    flow_request = CharField(source="flow_request.slug", required=False)

    default_locale = CharField(read_only=True)
    flags = SerializerMethodField()

    @extend_schema_field(field=PublicFlagsField)
    def get_flags(self, _):
        values = {}
        visibilities = ["public"]
        request = self.context.get("request")
        if request and request.user.is_authenticated:
            visibilities.append("authenticated")
        for visibility in visibilities:
            for flag in Flag.available(visibility=visibility):
                values[flag().key] = flag.get()
        return values

    def to_representation(self, instance: Brand) -> dict[str, Any]:
        data = super().to_representation(instance)
        # Suppress custom CSS for safe-mode sessions (e.g. recovery links) so that
        # misconfigured branding cannot prevent a user from reaching the UI to fix it.
        request = self.context.get("request")
        if request is not None and session_safe_mode(request):
            data["branding_custom_css"] = ""
        return data


class BrandViewSet(UsedByMixin, ModelViewSet):
    """Brand Viewset"""

    queryset = Brand.objects.all()
    serializer_class = BrandSerializer
    search_fields = [
        "domain",
        "branding_title",
        "web_certificate__name",
        "client_certificates__name",
    ]
    filterset_fields = [
        "brand_uuid",
        "domain",
        "default",
        "branding_title",
        "branding_logo",
        "branding_favicon",
        "branding_default_flow_background",
        "flow_authentication",
        "flow_user_switch",
        "flow_invalidation",
        "flow_recovery",
        "flow_unenrollment",
        "flow_user_settings",
        "flow_device_code",
        "flow_lockdown",
        "flow_request",
        "web_certificate",
        "client_certificates",
    ]
    ordering = ["domain"]

    filter_backends = [SecretKeyFilter, OrderingFilter, SearchFilter]

    @extend_schema(
        responses=CurrentBrandSerializer(many=False),
    )
    @action(methods=["GET"], detail=False, permission_classes=[AllowAny])
    def current(self, request: Request) -> Response:
        """Get current brand"""
        brand: Brand = request._request.brand
        return Response(CurrentBrandSerializer(brand, context={"request": request}).data)