TMSS-2305: deal with extra actions that create on other models and scheduling_set_id url parameter

SAS/TMSS/backend/src/tmss/tmssapp/viewsets/lofar_viewset.py

@@ -106,9 +106,13 @@ class LOFARViewSet(viewsets.ModelViewSet):
             for method in extra_action.mapping:
                 self.action = action  # pretend to do something else and check if we are allowed to do that
                 request.method = method.upper()  # pretend to do something else and check if we are allowed to do that
+                try:
                     if TMSSPermissions().has_permission(request=request, view=self):
                         if action not in allowed_methods:
                             allowed_methods.append(action)
+                except:
+                    pass
+                # todo: check extra actions that create on other models if allowerd when correct project is referenced
         self.action = actual_action
         request.method = actual_method
 

SAS/TMSS/backend/src/tmss/tmssapp/viewsets/permissions.py

@@ -24,6 +24,12 @@ User = get_user_model()
 #       A. apply a filter to prevent object without permission to be included in the list with full details, or
 #       B. customize get_queryset on the view to call check_object_permissions.
 
+# This dictionary maps extra actions on the model that should be considered for the project reference when checking
+# project-based permissions.
+# todo: discuss whether this reference should be added to the ProjectPermission table, so that we can populate it
+#  together with the required project roles.
+extra_action_project_related_model = {'create_scheduling_unit': models.SchedulingUnitDraft}
+
 def get_project_roles_for_user(user):
 
         if not isProductionEnvironment():
@@ -70,7 +76,7 @@ def get_project_roles_for_user(user):
 
 def get_project_roles_with_permission(permission_name, method='GET'):
     try:
-        logger.info('checking permission name=%s action=%s' % (permission_name, method))
+        logger.info('checking permission name=%s method=%s' % (permission_name, method))
         # ...retrieve ProjectPermission object
         project_permission = models.ProjectPermission.objects.get(name=permission_name)
         # ...determine what project roles are allowed to perform the requested action
@@ -154,20 +160,32 @@ class IsProjectMember(drf_permissions.DjangoObjectPermissions):
         # Note: the Viewflow actions are currently create actions but they do not actually create anything
         # (and do not contain the expected data to create an instance of the underlying model, but instead e.g. the
         # data required to perform a workflow step). So we exclude the offending views here.
-        if view.action == 'create' and request.data \
+        if (view.action == 'create' or view.action in extra_action_project_related_model) and request.data is not None\
                 and not 'SchedulingUnitTaskExecuteViewSet' in str(view) \
                 and not 'SchedulingUnitTaskAssignViewSet' in str(view):
             obj = None
-            if view.serializer_class.Meta.model == models.Project:
+
+            # Some extra actions create objects on other models. By default, we use the project reference of the base
+            # model of the action to check whether the user has one of the required roles in this project. On base models
+            # that are project-independent (e.g. SchedulingUnitObservingStrategyTemplate), we instead check the target
+            # model for the path to project.
+            model = extra_action_project_related_model.get(view.action, view.serializer_class.Meta.model)
+
+            if model == models.Project:
                 return False  # project creation solely depends on system role
-            if hasattr(view.serializer_class.Meta.model, 'path_to_project'):
-                attrs = view.serializer_class.Meta.model.path_to_project.split('__')
-            elif hasattr(view.serializer_class.Meta.model, 'project'):
+            if hasattr(model, 'path_to_project'):
+                attrs = model.path_to_project.split('__')
+            elif hasattr(model, 'project'):
                 attrs = ['project']
             else:
-                raise AttributeError(f'The model {view.serializer_class.Meta.model} requires an attribute "project" or "path_to_project" to check project permissions.')
+                raise AttributeError(f'The model {model} requires an attribute "project" or "path_to_project" to check project permissions.')
             for attr in attrs:
                 obj_resolved_in_this_iteration = False
+                if attr == 'scheduling_set' and 'scheduling_set_id' in request.query_params:
+                    # some extra actions do not carry all info in their POSTed data, but include references as url parameters
+                    from django.shortcuts import get_object_or_404
+                    obj = get_object_or_404(models.SchedulingSet, pk=request.query_params['scheduling_set_id'])
+                    continue
                 if not obj:
                     # on first iteration, the referenced object needs to be resolved from POSTed FQDN
                     obj_ref = request.data[attr]

SAS/TMSS/backend/test/t_permissions_project_roles.py

@@ -93,11 +93,15 @@ class ProjectPermissionTestCase(TestCase):
             su_template_url = client.get_schedulingunit_template("scheduling unit")['url']
 
         # user is shared_support
-        cls.scheduling_set_shared_support_url = cls.test_data_creator.post_data_and_get_url(cls.test_data_creator.SchedulingSet(project_url=cls.project_shared_support_url), '/scheduling_set/')
+        set = cls.test_data_creator.post_data_and_get_response_as_json_object(cls.test_data_creator.SchedulingSet(project_url=cls.project_shared_support_url), '/scheduling_set/')
+        cls.scheduling_set_shared_support_url = set['url']
+        cls.scheduling_set_shared_support_id = set['id']
         cls.scheduling_unit_draft_shared_support_url = cls.test_data_creator.post_data_and_get_url(cls.test_data_creator.SchedulingUnitDraft(template_url=su_template_url, scheduling_set_url=cls.scheduling_set_shared_support_url), '/scheduling_unit_draft/')
 
         # user is contact
-        cls.scheduling_set_contact_url = cls.test_data_creator.post_data_and_get_url(cls.test_data_creator.SchedulingSet(project_url=cls.project_contact_url), '/scheduling_set/')
+        set = cls.test_data_creator.post_data_and_get_response_as_json_object(cls.test_data_creator.SchedulingSet(project_url=cls.project_contact_url), '/scheduling_set/')
+        cls.scheduling_set_contact_url = set['url']
+        cls.scheduling_set_contact_id = set['id']
         cls.scheduling_unit_draft_contact_url = cls.test_data_creator.post_data_and_get_url(cls.test_data_creator.SchedulingUnitDraft(scheduling_set_url=cls.scheduling_set_contact_url), '/scheduling_unit_draft/')
 
         # user has no role
@@ -111,9 +115,12 @@ class ProjectPermissionTestCase(TestCase):
         cls.test_data_creator.post_data_and_get_url(cls.test_data_creator.ProjectPermission(name='taskdraft-create_task_blueprint', POST=[shared_support_role_url]), '/project_permission/')
         cls.test_data_creator.post_data_and_get_url(cls.test_data_creator.ProjectPermission(name='project', GET=[shared_support_role_url], POST=[]), '/project_permission/')
         cls.test_data_creator.post_data_and_get_url(cls.test_data_creator.ProjectPermission(name='project-my_roles', GET=[shared_support_role_url, friend_of_project_role_url]), '/project_permission/')
+        cls.test_data_creator.post_data_and_get_url(cls.test_data_creator.ProjectPermission(name='schedulingunitobservingstrategytemplate-create_scheduling_unit', GET=[shared_support_role_url], POST=[shared_support_role_url]), '/project_permission/')
 
         cls.task_template_url = cls.test_data_creator.post_data_and_get_url(cls.test_data_creator.TaskTemplate(), '/task_template/')
 
+        cls.scheduling_unit_observing_strategy_template_url = cls.test_data_creator.post_data_and_get_url(cls.test_data_creator.SchedulingUnitObservingStrategyTemplate(scheduling_unit_template_url=su_template_url), '/scheduling_unit_observing_strategy_template/')
+
         # create a new test_data_creator with the regular 'paulus' user, not superuser as in other tests, so that project permissions of tht user are checked.
         cls.auth = requests.auth.HTTPBasicAuth('paulus', 'pauluspass')
         cls.test_data_creator = TMSSRESTTestDataCreator(BASE_URL, cls.auth)
@@ -252,6 +259,37 @@ class ProjectPermissionTestCase(TestCase):
             self.assertEqual(r.status_code, 403)
             self.assertNotIn('Access-Control-Allow-Methods', r.headers)
 
+    # SchedulingUnitObservingStrategyTemplate.actions
+    
+    def test_strategy_template_create_SU_draft_raises_error_if_user_has_no_permission_for_related_project(self):
+        # create SU draft via the extra action on strategy template, to test that the path to project of the SU draft model is used
+        # make sure we cannot create a draft in a scheduling set connected to a project where we have no role
+        POST_and_assert_expected_response(self, self.scheduling_unit_observing_strategy_template_url + '/create_scheduling_unit/',
+                                          {}, 403, {}, auth=self.auth, params={'scheduling_set_id': self.scheduling_set_contact_id})
+
+    def test_strategy_template_create_SU_draft_works_if_user_has_permission_for_related_project(self):
+        # create SU draft via the extra action on strategy template, to test that the path to project of the SU draft model is used
+        # make sure we can create a draft in a scheduling set connected to a project where we have 'shared_support' role
+        POST_and_assert_expected_response(self, self.scheduling_unit_observing_strategy_template_url + '/create_scheduling_unit/',
+                                          {}, 201, {}, auth=self.auth, params={'scheduling_set_id': self.scheduling_set_shared_support_id})
+
+    @unittest.skip('fix LOFARViewSet with check if extra action can be executed when correct project is referenced')
+    def test_access_control_allow_header_reflects_user_permissions_in_strategy_template_list_view(self):
+        # the frontend relies on the list view to include an extra action if the user has permission to execute it.
+        # Since the user has permission to create scheduling units in case they belong to the right project, the action
+        # should be listed here.
+        with requests.Session() as session:
+            session.verify = False
+            session.auth = self.auth
+
+            r = session.get(BASE_URL + '/scheduling_unit_observing_strategy_template/')
+            self.assertEqual(r.status_code, 200)
+            allowed_methods = r.headers['Access-Control-Allow-Methods'].split(', ')
+            self.assertIn('create_scheduling_unit', allowed_methods)
+
+
+    # Project
+
     def test_project_get_friend_returns_correct_user(self):
         """
         Note: This test relies on real data from Keycloak.

SAS/TMSS/backend/test/tmss_test_environment_unittest_setup.py

@@ -58,22 +58,22 @@ import lofar.sas.tmss.tmss.settings as TMSS_SETTINGS
 # by default we assert on requests taking longer than this timeout
 DEFAULT_REQUEST_TIMEOUT=10
 
-def _call_API_and_assert_expected_response(test_instance, url, call, data, expected_code, expected_content, auth=AUTH, timeout:float=DEFAULT_REQUEST_TIMEOUT):
+def _call_API_and_assert_expected_response(test_instance, url, call, data, expected_code, expected_content, auth=AUTH, timeout:float=DEFAULT_REQUEST_TIMEOUT, params=None):
     """
     Call API method on the provided url and assert the expected code is returned and the expected content is in the response content
     :return: response as dict. This either contains the data of an entry or error details. If JSON cannot be parsed, return string.
     """
     _start_request_timestamp = datetime.datetime.utcnow()
     if call == 'PUT':
-        response = requests.put(url, json=data, auth=auth, timeout=timeout)
+        response = requests.put(url, json=data, auth=auth, timeout=timeout, params=params)
     elif call == 'POST':
-        response = requests.post(url, json=data, auth=auth, timeout=timeout)
+        response = requests.post(url, json=data, auth=auth, timeout=timeout, params=params)
     elif call == 'GET':
-        response = requests.get(url, auth=auth, allow_redirects=False)
+        response = requests.get(url, auth=auth, allow_redirects=False, params=params)
     elif call == 'PATCH':
-        response = requests.patch(url, json=data, auth=auth, timeout=timeout)
+        response = requests.patch(url, json=data, auth=auth, timeout=timeout, params=params)
     elif call == 'DELETE':
-        response = requests.delete(url, auth=auth, timeout=timeout)
+        response = requests.delete(url, auth=auth, timeout=timeout, params=params)
     else:
         raise ValueError("The provided call '%s' is not a valid API method choice" % call)
 
@@ -131,12 +131,12 @@ def PUT_and_assert_expected_response(test_instance, url, data, expected_code, ex
     return r_dict
 
 
-def POST_and_assert_expected_response(test_instance, url, data, expected_code, expected_content, auth=AUTH, timeout:float=DEFAULT_REQUEST_TIMEOUT):
+def POST_and_assert_expected_response(test_instance, url, data, expected_code, expected_content, auth=AUTH, timeout:float=DEFAULT_REQUEST_TIMEOUT, params=None):
     """
     POST data on url and assert the expected code is returned and the expected content is in the response content
     :return: response dict
     """
-    r_dict = _call_API_and_assert_expected_response(test_instance, url, 'POST', data, expected_code, expected_content, auth=auth, timeout=timeout)
+    r_dict = _call_API_and_assert_expected_response(test_instance, url, 'POST', data, expected_code, expected_content, auth=auth, timeout=timeout, params=params)
     return r_dict