from rest_framework import status from collaborators import defined_roles as dr from collaborators.defined_roles import role_permission_check from collaborators.models import Access from exam.models import ExamPage, ExamTemplate, Exam from institute.models import Institute, APIKey from question.models import Question from question_bank.models import QuestionBank from quiz.models import Quiz, QuizTemplate, QuizSheet from text_book.models import AnswerSheet from utils.myresponse import MyResponse class CollaborationAccessMiddleware: """ This middleware class is responsible for checking user access permission. if it was successful it passes the request, otherwise it responses the user with an access denied error! """ def __init__(self, get_response, permissions, model=None, db_field_name=None, serializer_field_name=None): self.get_response = get_response self.permissions = permissions self.model = model self.db_field_name = db_field_name self.serializer_field_name = serializer_field_name # One-time configuration and initialization. def __call__(self, request): # Code to be executed for each request before # the view (and later middleware) are called. response = self.get_response(request) # Code to be executed for each request/response after # the view is called. return response def process_view(self, request, view_func, view_args, view_kwargs): # ########################## Block 1 ################################ # permission check process for user request if hasattr(request, 'middleware_user'): # ########################## Block 1-1 ################################ # get serializer_field_value and try to get appropriate record from model db # TODO sometimes we cannot put institute_id in the serializer e.g. ListExamRule serializer_field_value = request.middleware_serializer_data.get( self.serializer_field_name) # TODO check for query injection by json object injection query_dict = {self.db_field_name: serializer_field_value} try: model_record = self.model.objects.get(**query_dict) except: return MyResponse(request, {'status': 'error', 'message': 'Forbidden'}, status=status.HTTP_403_FORBIDDEN) # ########################## Block 1-2 ################################ # determine institute based on model and checks if the user of institute is as same as middleware_user institute = None if self.model == Institute: institute = model_record elif self.model == APIKey: institute = model_record.institute elif self.model == QuestionBank: institute = model_record.institute elif self.model == Question: institute = model_record.question_bank.institute elif self.model == ExamTemplate: institute = model_record.institute elif self.model == Exam: institute = model_record.institute elif self.model == AnswerSheet: institute = model_record.question.question_bank.institute elif self.model == ExamPage: institute = model_record.institute elif self.model == Quiz: institute = model_record.institute elif self.model == QuizSheet: institute = model_record.quiz.institute elif self.model == QuizTemplate: institute = model_record.institute request.middleware_institute = institute request.middleware_model_record = model_record # ########################## Block 1-3 ################################ # This section of the code is no longer useful after CheckNoneCollaborativeReadOnlyAccessMiddleware was added # # 70274207 test.py # # [__PERMISSION_READ_QUESTIONBANK__, __PERMISSION_READ_QUESTION__] permissions are granted on public question banks # if self.model == QuestionBank: # if model_record.public: # public_permission_denied = False # for permission in self.permissions: # if permission not in [dr.__PERMISSION_READ_QUESTIONBANK__, dr.__PERMISSION_READ_QUESTION__]: # public_permission_denied = True # break # if not public_permission_denied: # return # # Also, [__PERMISSION_READ_QUESTIONBANK__, __PERMISSION_READ_QUESTION__] permissions are granted # # for questions with public question_bank # elif self.model == Question: # if model_record.question_bank.public: # public_permission_denied = False # for permission in self.permissions: # if permission not in [dr.__PERMISSION_READ_QUESTIONBANK__, dr.__PERMISSION_READ_QUESTION__]: # public_permission_denied = True # break # if not public_permission_denied: # return # # # if public permission was not granted, # try to get Access from db based on institute and middleware_user and checks the permissions try: access_record = Access.objects.get( user=request.middleware_user, institute=institute) request.middleware_access = access_record except: return MyResponse(request, {'status': 'error', 'message': 'Forbidden'}, status=status.HTTP_403_FORBIDDEN) roles = access_record.roles for permission in self.permissions: if not role_permission_check(roles, permission): return MyResponse(request, {'status': 'error', 'message': 'Forbidden'}, status=status.HTTP_403_FORBIDDEN) # ########################## Block 2 ################################ # permission check process for server request elif hasattr(request, 'middleware_api_key'): for permission in self.permissions: if not role_permission_check(request.middleware_api_key.roles, permission): return MyResponse(request, {'status': 'error', 'message': 'Forbidden'}, status=status.HTTP_403_FORBIDDEN) # ########################## Block 3 ################################ # return error if request was sent from none of them else: return MyResponse(request, {'status': 'error', 'message': 'Bad Request'}, status=status.HTTP_400_BAD_REQUEST) class SubscriptionAccessMiddleware: """ This middleware checks if the user (representative of context_institute_id) has owner or subscription or ... accessibility or not """ def __init__(self, get_response, model=None, db_field_name=None, serializer_field_name=None): self.get_response = get_response self.model = model self.db_field_name = db_field_name self.serializer_field_name = serializer_field_name # One-time configuration and initialization. def __call__(self, request): # Code to be executed for each request before # the view (and later middleware) are called. response = self.get_response(request) # Code to be executed for each request/response after # the view is called. return response def process_view(self, request, view_func, view_args, view_kwargs): # ########################## Block 1 ################################ # permission check process for user request if hasattr(request, 'middleware_institute'): # ########################## Block 1-1 ################################ serializer_field_value = request.middleware_serializer_data.get( self.serializer_field_name) query_dict = {self.db_field_name: serializer_field_value} try: model_record = self.model.objects.get(**query_dict) except: return MyResponse(request, {'status': 'error', 'message': 'Forbidden'}, status=status.HTTP_403_FORBIDDEN) # ########################## Block 1-2 ################################ if self.model == QuestionBank: request.middleware_question_bank = model_record elif self.model == Question: request.middleware_question_bank = model_record.question_bank request.middleware_question = model_record else: return MyResponse(request, {'status': 'error', 'message': 'Bad Request'}, status=status.HTTP_400_BAD_REQUEST) context_institute = request.middleware_institute if context_institute.can_read_question_bank(request.middleware_question_bank): return else: return MyResponse(request, {'status': 'error', 'message': 'Forbidden'}, status=status.HTTP_403_FORBIDDEN) # ########################## Block 3 ################################ # return error if request was sent from none of them else: return MyResponse(request, {'status': 'error', 'message': 'Bad Request'}, status=status.HTTP_400_BAD_REQUEST)