2bcbac8454
Related to https://github.com/grafana/oncall-private/issues/2826 Continuing work started in https://github.com/grafana/oncall/pull/5211, this adds support for Grafana service accounts tokens for API authentication (except alert group actions which will still require a user behind). Next steps would be updating the go client and the terraform provider to allow service account token auth for OnCall resources. Following proposal 1.1 from [doc](https://docs.google.com/document/d/1I3nFbsUEkiNPphBXT-kWefIeramTY71qqZ1OA06Kmls/edit?usp=sharing).
100 lines
3.8 KiB
Python
100 lines
3.8 KiB
Python
from django_filters import rest_framework as filters
|
|
from rest_framework.decorators import action
|
|
from rest_framework.exceptions import NotFound
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from rest_framework.views import Response
|
|
from rest_framework.viewsets import ReadOnlyModelViewSet
|
|
|
|
from apps.api.permissions import LegacyAccessControlRole, RBACPermission
|
|
from apps.auth_token.auth import (
|
|
ApiTokenAuthentication,
|
|
GrafanaServiceAccountAuthentication,
|
|
UserScheduleExportAuthentication,
|
|
)
|
|
from apps.public_api.custom_renderers import CalendarRenderer
|
|
from apps.public_api.serializers import FastUserSerializer, UserSerializer
|
|
from apps.public_api.tf_sync import is_request_from_terraform, sync_users_on_tf_request
|
|
from apps.public_api.throttlers.user_throttle import UserThrottle
|
|
from apps.schedules.ical_utils import user_ical_export
|
|
from apps.schedules.models import OnCallSchedule
|
|
from apps.user_management.models import User
|
|
from common.api_helpers.mixins import RateLimitHeadersMixin, ShortSerializerMixin
|
|
from common.api_helpers.paginators import HundredPageSizePaginator
|
|
|
|
|
|
class UserFilter(filters.FilterSet):
|
|
"""
|
|
https://django-filter.readthedocs.io/en/master/guide/rest_framework.html
|
|
"""
|
|
|
|
email = filters.CharFilter(field_name="email", lookup_expr="iexact")
|
|
roles = filters.MultipleChoiceFilter(
|
|
field_name="role", choices=LegacyAccessControlRole.choices()
|
|
) # LEGACY, should be removed eventually
|
|
username = filters.CharFilter(field_name="username", lookup_expr="iexact")
|
|
|
|
class Meta:
|
|
model = User
|
|
fields = ["email", "roles", "username"]
|
|
|
|
|
|
class UserView(RateLimitHeadersMixin, ShortSerializerMixin, ReadOnlyModelViewSet):
|
|
authentication_classes = (GrafanaServiceAccountAuthentication, ApiTokenAuthentication)
|
|
permission_classes = (IsAuthenticated, RBACPermission)
|
|
|
|
rbac_permissions = {
|
|
"list": [RBACPermission.Permissions.USER_SETTINGS_READ],
|
|
"retrieve": [RBACPermission.Permissions.USER_SETTINGS_READ],
|
|
}
|
|
|
|
model = User
|
|
pagination_class = HundredPageSizePaginator
|
|
|
|
serializer_class = UserSerializer
|
|
short_serializer_class = FastUserSerializer
|
|
filterset_class = UserFilter
|
|
filter_backends = (filters.DjangoFilterBackend,)
|
|
|
|
throttle_classes = [UserThrottle]
|
|
|
|
# self.get_object() is not used in export action because UserScheduleExportAuthentication is used
|
|
extra_actions_ignore_no_get_object = ["schedule_export"]
|
|
|
|
def get_queryset(self):
|
|
if is_request_from_terraform(self.request):
|
|
sync_users_on_tf_request(self.request.auth.organization)
|
|
is_short_request = self.request.query_params.get("short", "false") == "true"
|
|
queryset = self.request.auth.organization.users.all()
|
|
if not is_short_request:
|
|
queryset = self.serializer_class.setup_eager_loading(queryset)
|
|
queryset = self.filter_queryset(queryset)
|
|
return queryset.order_by("id")
|
|
|
|
def get_object(self):
|
|
public_primary_key = self.kwargs["pk"]
|
|
|
|
if public_primary_key == "current":
|
|
return self.request.user
|
|
|
|
organization = self.request.auth.organization
|
|
|
|
try:
|
|
user = User.objects.get(public_primary_key=public_primary_key, organization=organization)
|
|
except User.DoesNotExist:
|
|
raise NotFound
|
|
|
|
return user
|
|
|
|
@action(
|
|
methods=["get"],
|
|
detail=True,
|
|
renderer_classes=(CalendarRenderer,),
|
|
authentication_classes=(UserScheduleExportAuthentication,),
|
|
permission_classes=(IsAuthenticated,),
|
|
)
|
|
def schedule_export(self, request, pk):
|
|
schedules = OnCallSchedule.objects.filter(organization=self.request.auth.organization).related_to_user(
|
|
self.request.user
|
|
)
|
|
export = user_ical_export(self.request.user, schedules)
|
|
return Response(export)
|