trustchain-oc1-orchestral
/
authentik
Archived
10
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

core: only cache Applications API when no filtering is done

This commit is contained in:
Jens Langhammer 2021-01-28 23:16:51 +01:00
parent 5ef4354723
commit 188ef0f58f
1 changed files with 28 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
authentik/core/api/applications.py
View File

@ -11,6 +11,7 @@ from rest_framework.response import Response
from rest_framework.serializers import ModelSerializer from rest_framework.serializers import ModelSerializer
from rest_framework.viewsets import ModelViewSet from rest_framework.viewsets import ModelViewSet
from rest_framework_guardian.filters import ObjectPermissionsFilter from rest_framework_guardian.filters import ObjectPermissionsFilter
from structlog.stdlib import get_logger
from authentik.admin.api.metrics import get_events_per_1h from authentik.admin.api.metrics import get_events_per_1h
from authentik.core.api.providers import ProviderSerializer from authentik.core.api.providers import ProviderSerializer
@ -18,6 +19,8 @@ from authentik.core.models import Application
from authentik.events.models import EventAction from authentik.events.models import EventAction
from authentik.policies.engine import PolicyEngine from authentik.policies.engine import PolicyEngine
LOGGER = get_logger()
def user_app_cache_key(user_pk: str) -> str: def user_app_cache_key(user_pk: str) -> str:
"""Cache key where application list for user is saved""" """Cache key where application list for user is saved"""
@ -74,18 +77,30 @@ class ApplicationViewSet(ModelViewSet):
queryset = backend().filter_queryset(self.request, queryset, self) queryset = backend().filter_queryset(self.request, queryset, self)
return queryset return queryset
def list(self, request: Request) -> Response: def _get_allowed_applications(self, queryset: QuerySet) -> list[Application]:
"""Custom list method that checks Policy based access instead of guardian""" applications = []
queryset = self._filter_queryset_for_list(self.get_queryset())
self.paginate_queryset(queryset)
allowed_applications = cache.get(user_app_cache_key(self.request.user.pk))
if not allowed_applications:
allowed_applications = []
for application in queryset: for application in queryset:
engine = PolicyEngine(application, self.request.user, self.request) engine = PolicyEngine(application, self.request.user, self.request)
engine.build() engine.build()
if engine.passing: if engine.passing:
allowed_applications.append(application) applications.append(application)
return applications
def list(self, request: Request) -> Response:
"""Custom list method that checks Policy based access instead of guardian"""
queryset = self._filter_queryset_for_list(self.get_queryset())
self.paginate_queryset(queryset)
should_cache = "search" not in request.GET
allowed_applications = []
if not should_cache:
allowed_applications = self._get_allowed_applications(queryset)
if should_cache:
LOGGER.debug("Caching allowed application list")
allowed_applications = cache.get(user_app_cache_key(self.request.user.pk))
if not allowed_applications:
allowed_applications = self._get_allowed_applications(queryset)
cache.set( cache.set(
user_app_cache_key(self.request.user.pk), user_app_cache_key(self.request.user.pk),
allowed_applications, allowed_applications,