authentik.rbac.filters

  1"""RBAC API Filter"""
  2
  3from django.conf import settings
  4from django.db.models import QuerySet
  5from django_filters.rest_framework import DjangoFilterBackend
  6from rest_framework.authentication import get_authorization_header
  7from rest_framework.exceptions import PermissionDenied
  8from rest_framework.filters import BaseFilterBackend
  9from rest_framework.request import Request
 10from rest_framework.views import APIView
 11
 12from authentik.api.authentication import validate_auth
 13from authentik.core.models import UserTypes
 14
 15
 16# Inline fork of https://github.com/rpkilby/django-rest-framework-guardian
 17# BSD 3-Clause License
 18#
 19# Copyright (c) 2018, Ryan P Kilby
 20# All rights reserved.
 21#
 22# Redistribution and use in source and binary forms, with or without
 23# modification, are permitted provided that the following conditions are met:
 24#
 25# * Redistributions of source code must retain the above copyright notice, this
 26#   list of conditions and the following disclaimer.
 27#
 28# * Redistributions in binary form must reproduce the above copyright notice,
 29#   this list of conditions and the following disclaimer in the documentation
 30#   and/or other materials provided with the distribution.
 31#
 32# * Neither the name of the copyright holder nor the names of its
 33#   contributors may be used to endorse or promote products derived from
 34#   this software without specific prior written permission.
 35#
 36# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 37# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 38# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 39# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
 40# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 41# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 42# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
 43# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 44# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 45# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 46class ObjectPermissionsFilter(BaseFilterBackend):
 47    """
 48    A filter backend that limits results to those where the requesting user
 49    has read object level permissions.
 50    """
 51
 52    perm_format = "%(app_label)s.view_%(model_name)s"
 53
 54    def filter_queryset(self, request, queryset, view):
 55        # We want to defer this import until runtime, rather than import-time.
 56        # See https://github.com/encode/django-rest-framework/issues/4608
 57        # (Also see #1624 for why we need to make this import explicitly)
 58        from guardian.shortcuts import get_objects_for_user
 59
 60        user = request.user
 61        permission = self.perm_format % {
 62            "app_label": queryset.model._meta.app_label,
 63            "model_name": queryset.model._meta.model_name,
 64        }
 65
 66        return get_objects_for_user(user, permission, queryset)
 67
 68
 69class ObjectFilter(ObjectPermissionsFilter):
 70    """Object permission filter that grants global permission higher priority than
 71    per-object permissions"""
 72
 73    def filter_queryset(self, request: Request, queryset: QuerySet, view: APIView) -> QuerySet:
 74        permission = self.perm_format % {
 75            "app_label": queryset.model._meta.app_label,
 76            "model_name": queryset.model._meta.model_name,
 77        }
 78        # having the global permission set on a user has higher priority than
 79        # per-object permissions
 80        if request.user.has_perm(permission):
 81            return queryset
 82        # User does not have permissions, but we have an owner field defined, so filter by that
 83        if owner_field := getattr(view, "owner_field", None):
 84            return queryset.filter(**{owner_field: request.user})
 85        queryset = super().filter_queryset(request, queryset, view)
 86        # Outposts (which are the only objects using internal service accounts)
 87        # except requests to return an empty list when they have no objects
 88        # assigned
 89        if getattr(request.user, "type", None) == UserTypes.INTERNAL_SERVICE_ACCOUNT:
 90            return queryset
 91        if not queryset.exists():
 92            # User doesn't have direct permission to all objects
 93            # and also no object permissions assigned (directly or via role)
 94            raise PermissionDenied()
 95        return queryset
 96
 97
 98class SecretKeyFilter(DjangoFilterBackend):
 99    """Allow access to all objects when authenticated with secret key as token.
100
101    Replaces both DjangoFilterBackend and ObjectFilter"""
102
103    def filter_queryset(self, request: Request, queryset: QuerySet, view) -> QuerySet:
104        auth_header = get_authorization_header(request)
105        token = validate_auth(auth_header)
106        if token and token == settings.SECRET_KEY:
107            return queryset
108        queryset = ObjectFilter().filter_queryset(request, queryset, view)
109        return super().filter_queryset(request, queryset, view)