authentik.sources.scim.views.v2.base

  1"""SCIM Utils"""
  2
  3from typing import Any
  4from uuid import UUID
  5
  6from django.core.paginator import Page, Paginator
  7from django.db.models import Q, QuerySet
  8from django.http import HttpRequest
  9from rest_framework.parsers import JSONParser
 10from rest_framework.permissions import IsAuthenticated
 11from rest_framework.renderers import JSONRenderer
 12from rest_framework.request import Request
 13from rest_framework.response import Response
 14from rest_framework.views import APIView
 15from scim2_filter_parser.transpilers.django_q_object import get_query
 16from structlog import BoundLogger
 17from structlog.stdlib import get_logger
 18
 19from authentik.core.models import Group, User
 20from authentik.core.sources.mapper import SourceMapper
 21from authentik.lib.sync.mapper import PropertyMappingManager
 22from authentik.sources.scim.models import SCIMSource
 23from authentik.sources.scim.views.v2.auth import SCIMTokenAuth
 24from authentik.sources.scim.views.v2.exceptions import SCIMNotFoundError
 25
 26SCIM_CONTENT_TYPE = "application/scim+json"
 27
 28
 29class SCIMParser(JSONParser):
 30    """SCIM clients use a custom content type"""
 31
 32    media_type = SCIM_CONTENT_TYPE
 33
 34
 35class SCIMRenderer(JSONRenderer):
 36    """SCIM clients also expect a custom content type"""
 37
 38    media_type = SCIM_CONTENT_TYPE
 39
 40
 41class SCIMView(APIView):
 42    """Base class for SCIM Views"""
 43
 44    source: SCIMSource
 45    logger: BoundLogger
 46
 47    permission_classes = [IsAuthenticated]
 48    parser_classes = [SCIMParser, JSONParser]
 49    renderer_classes = [SCIMRenderer]
 50
 51    def setup(self, request: HttpRequest, *args: Any, **kwargs: Any) -> None:
 52        self.logger = get_logger().bind()
 53        super().setup(request, *args, **kwargs)
 54
 55    def get_authenticators(self):
 56        return [SCIMTokenAuth(self)]
 57
 58    def remove_excluded_attributes(self, data: dict):
 59        """Remove attributes specified in excludedAttributes"""
 60        excluded: str = self.request.query_params.get("excludedAttributes", "")
 61        for key in excluded.split(","):
 62            data.pop(key.strip(), None)
 63        return data
 64
 65    def filter_parse(self, request: Request):
 66        """Parse the path of a Patch Operation"""
 67        path = request.query_params.get("filter")
 68        if not path:
 69            return Q()
 70        attr_map = {}
 71        if self.model == User:
 72            attr_map = {
 73                ("userName", None, None): "user__username",
 74                ("active", None, None): "user__is_active",
 75                ("name", "familyName", None): "attributes__familyName",
 76            }
 77        elif self.model == Group:
 78            attr_map = {
 79                ("displayName", None, None): "group__name",
 80                ("members", None, None): "group__users",
 81            }
 82        return get_query(
 83            path,
 84            attr_map,
 85        )
 86
 87    def paginate_query(self, query: QuerySet) -> Page:
 88        per_page = int(self.request.tenant.pagination_default_page_size)
 89        start_index = 1
 90        try:
 91            start_index = int(self.request.query_params.get("startIndex", 1))
 92        except ValueError:
 93            pass
 94        paginator = Paginator(query, per_page=per_page)
 95        page = paginator.page(int(max(start_index / per_page, 1)))
 96        return page
 97
 98
 99class SCIMObjectView(SCIMView):
100    """Base SCIM View for object management"""
101
102    mapper: SourceMapper
103    manager: PropertyMappingManager
104
105    model: type[User | Group]
106
107    def initial(self, request: Request, *args, **kwargs) -> None:
108        super().initial(request, *args, **kwargs)
109        # This needs to happen after authentication has happened, because we don't have
110        # a source attribute before
111        self.mapper = SourceMapper(self.source)
112        self.manager = self.mapper.get_manager(self.model, ["data"])
113        for key, value in kwargs.items():
114            if key.endswith("_id"):
115                try:
116                    UUID(value)
117                except ValueError:
118                    raise SCIMNotFoundError("Invalid ID") from None
119
120    def build_object_properties(self, data: dict[str, Any]) -> dict[str, Any | dict[str, Any]]:
121        return self.mapper.build_object_properties(
122            object_type=self.model,
123            manager=self.manager,
124            user=None,
125            request=self.request,
126            data=data,
127        )
128
129
130class SCIMRootView(SCIMView):
131    """Root SCIM View"""
132
133    def dispatch(self, request: Request, *args, **kwargs) -> Response:
134        return Response({"message": "Use this base-URL with a SCIM-compatible system."})