ServiceAccess tests + adjust 400->403 for forbidden children resource…

… creation (relates to #359)

CHANGES.rst

@@ -56,6 +56,8 @@ Features / Changes
 Bug Fixes
 ~~~~~~~~~~~~~~~~~~~~~
 * Fix incorrect regex employed for validation of service URL during registration.
+* Replace HTTP status code ``400`` by ``403`` and ``422`` where applicable for invalid resource creation due to failing
+  validations against reference parent service (relates to `#359 <https://github.com/Ouranosinc/Magpie/issues/359>`_).
 
 `2.0.1 <https://github.com/Ouranosinc/Magpie/tree/2.0.1>`_ (2020-09-30)
 ------------------------------------------------------------------------------------

magpie/api/management/resource/resource_utils.py

@@ -7,7 +7,8 @@
     HTTPForbidden,
     HTTPInternalServerError,
     HTTPNotFound,
-    HTTPOk
+    HTTPOk,
+    HTTPUnprocessableEntity
 )
 from pyramid.settings import asbool
 from ziggurat_foundations.models.services.resource import ResourceService
@@ -69,17 +70,17 @@ def check_valid_service_resource(parent_resource, resource_type, db_session):
     parent_type = parent_resource.resource_type_name
     parent_msg_err = "Child resource not allowed for specified parent resource type '{}'".format(parent_type)
     ax.verify_param(models.RESOURCE_TYPE_DICT[parent_type].child_resource_allowed, is_equal=True,
-                    param_compare=True, http_error=HTTPBadRequest, msg_on_fail=parent_msg_err)
+                    param_compare=True, http_error=HTTPForbidden, msg_on_fail=parent_msg_err)
     root_service = get_resource_root_service(parent_resource, db_session=db_session)
     ax.verify_param(root_service, not_none=True, http_error=HTTPInternalServerError,
                     msg_on_fail="Failed retrieving 'root_service' from db")
     ax.verify_param(root_service.resource_type, is_equal=True, http_error=HTTPInternalServerError,
                     param_name="resource_type", param_compare=models.Service.resource_type_name,
                     msg_on_fail="Invalid 'root_service' retrieved from db is not a service")
     ax.verify_param(SERVICE_TYPE_DICT[root_service.type].child_resource_allowed, is_equal=True,
-                    param_compare=True, http_error=HTTPBadRequest,
+                    param_compare=True, http_error=HTTPForbidden,
                     msg_on_fail="Child resource not allowed for specified service type '{}'".format(root_service.type))
-    ax.verify_param(resource_type, is_in=True, http_error=HTTPBadRequest,
+    ax.verify_param(resource_type, is_in=True, http_error=HTTPForbidden,
                     param_name="resource_type", param_compare=SERVICE_TYPE_DICT[root_service.type].resource_type_names,
                     msg_on_fail="Invalid 'resource_type' specified for service type '{}'".format(root_service.type))
     return root_service
@@ -271,17 +272,18 @@ def get_resource_root_service_impl(resource, request):
 def create_resource(resource_name, resource_display_name, resource_type, parent_id, db_session):
     # type: (Str, Optional[Str], Str, int, Session) -> HTTPException
     ax.verify_param(resource_name, param_name="resource_name", not_none=True, not_empty=True,
-                    http_error=HTTPBadRequest,
+                    http_error=HTTPUnprocessableEntity,
                     msg_on_fail="Invalid 'resource_name' specified for child resource creation.")
     ax.verify_param(resource_type, param_name="resource_type", not_none=True, not_empty=True,
-                    http_error=HTTPBadRequest,
+                    http_error=HTTPUnprocessableEntity,
                     msg_on_fail="Invalid 'resource_type' specified for child resource creation.")
     ax.verify_param(parent_id, param_name="parent_id", not_none=True, is_type=True, param_compare=int,
-                    http_error=HTTPBadRequest, msg_on_fail="Invalid 'parent_id' specified for child resource creation.")
+                    http_error=HTTPUnprocessableEntity,
+                    msg_on_fail="Invalid 'parent_id' specified for child resource creation.")
     parent_resource = ax.evaluate_call(lambda: ResourceService.by_resource_id(parent_id, db_session=db_session),
                                        fallback=lambda: db_session.rollback(), http_error=HTTPNotFound,
                                        msg_on_fail=s.Resources_POST_NotFoundResponseSchema.description,
-                                       content={"parent_id": str(parent_id), "resource_name": str(resource_name),
+                                       content={"parent_id": parent_id, "resource_name": str(resource_name),
                                                 "resource_type": str(resource_type)})
 
     # verify for valid permissions from top-level service-specific corresponding resources permissions

magpie/api/schemas.py

@@ -1289,12 +1289,11 @@ class ServiceResources_POST_RequestSchema(Resources_POST_RequestSchema):
 
 
 ServiceResources_POST_CreatedResponseSchema = Resources_POST_CreatedResponseSchema
-ServiceResources_POST_ForbiddenResponseSchema = Resources_POST_ForbiddenResponseSchema
 ServiceResources_POST_NotFoundResponseSchema = Resources_POST_NotFoundResponseSchema
 ServiceResources_POST_ConflictResponseSchema = Resources_POST_ConflictResponseSchema
 
 
-class ServiceResources_POST_BadRequestResponseSchema(Resources_POST_BadRequestResponseSchema):
+class ServiceResources_POST_ForbiddenResponseSchema(Resources_POST_ForbiddenResponseSchema):
     description = "Invalid 'parent_id' specified for child resource creation under requested service."
 
 
@@ -2727,9 +2726,8 @@ class SwaggerAPI_GET_OkResponseSchema(colander.MappingSchema):
 }
 Resources_POST_responses = {
     "201": Resources_POST_CreatedResponseSchema(),
-    "400": Resources_POST_BadRequestResponseSchema(),  # FIXME: https://github.com/Ouranosinc/Magpie/issues/359
     "401": UnauthorizedResponseSchema(),
-    "403": Resources_POST_ForbiddenResponseSchema(),  # FIXME: https://github.com/Ouranosinc/Magpie/issues/359
+    "403": Resources_POST_ForbiddenResponseSchema(),
     "404": Resources_POST_NotFoundResponseSchema(),
     "406": NotAcceptableResponseSchema(),
     "409": Resources_POST_ConflictResponseSchema(),
@@ -2832,9 +2830,8 @@ class SwaggerAPI_GET_OkResponseSchema(colander.MappingSchema):
 }
 ServiceResources_POST_responses = {
     "201": ServiceResources_POST_CreatedResponseSchema(),
-    "400": ServiceResources_POST_BadRequestResponseSchema(),  # FIXME: https://github.com/Ouranosinc/Magpie/issues/359
     "401": UnauthorizedResponseSchema(),
-    "403": ServiceResources_POST_ForbiddenResponseSchema(),  # FIXME: https://github.com/Ouranosinc/Magpie/issues/359
+    "403": ServiceResources_POST_ForbiddenResponseSchema(),
     "404": ServiceResources_POST_NotFoundResponseSchema(),
     "406": NotAcceptableResponseSchema(),
     "409": ServiceResources_POST_ConflictResponseSchema(),

magpie/models.py

@@ -304,6 +304,7 @@ class Route(Resource):
 
 
 class Process(Resource):
+    child_resource_allowed = False
     resource_type_name = "process"
     __mapper_args__ = {"polymorphic_identity": resource_type_name}
 

magpie/permissions.py

@@ -2,6 +2,7 @@
 import itertools
 from typing import TYPE_CHECKING
 
+import six
 from pyramid.security import ALL_PERMISSIONS, Everyone
 
 from magpie.utils import ExtendedEnum
@@ -320,7 +321,7 @@ def convert(cls, permission):
             # if permission name represents explicit definition, use it directly and drop Allow/Deny from ACE
             # otherwise, use the provided access
             access = None
-            if len(perm_name.split("-")) != 3:
+            if isinstance(perm_name, six.string_types) and len(perm_name.split("-")) != 3:
                 access = Access.get(permission[0].lower())
             return PermissionSet(perm_name, access=access, scope=None, typ=perm_type)
 

tests/interfaces.py

@@ -1476,12 +1476,18 @@ def test_GetLoggedUserResourcesPermissions_Queries(self):
         """
         Validate returned logged user permissions with different query parameter modifiers.
 
+        .. note::
+            With later versions, effective permissions are evaluated by
+            :meth:`test_GetUserResourcePermissions_EffectiveResolution` instead of within this test as rules require
+            more advanced validations and conditions with the addition of :class:`Access` and :class:`Scope` concepts.
+
         .. seealso::
             - :meth:`Interface_MagpieAPI_AdminAuth.test_GetLoggedUserResourcesPermissions`
+            - :meth:`Interface_MagpieAPI_AdminAuth.test_GetUserResourcePermissions_EffectiveResolution`
             - :meth:`Interface_MagpieAPI_UsersAuth.test_GetUserResourcesPermissions_AllowedItself`
             - :meth:`Interface_MagpieAPI_UsersAuth.test_GetUserResourcesPermissions_ForbiddenOther`
         """
-        utils.warn_version(self, "permission effect queries", "0.7.0", skip=True)
+        utils.warn_version(self, "permission queries modifiers", "0.7.0", skip=True)
 
         # setup test resources under service with permissions
         # Service/Resources              | Admin-User | Admin-Group | Anonym-User | Anonym-Group
@@ -1526,43 +1532,50 @@ def test_GetLoggedUserResourcesPermissions_Queries(self):
         # tests
         q_groups = "inherit=true"
         q_effect = "effective=true"
+        test_effective = LooseVersion(self.version) < LooseVersion("2.1")
         body = self.check_GetUserResourcesPermissions(self.usr, resource_id=test_child_res_id, query=None)
         utils.check_val_equal(body["permission_names"], [])
         body = self.check_GetUserResourcesPermissions(self.usr, resource_id=test_child_res_id, query=q_groups)
         utils.check_val_equal(body["permission_names"], [])
-        body = self.check_GetUserResourcesPermissions(self.usr, resource_id=test_child_res_id, query=q_effect)
-        utils.check_all_equal(body["permission_names"], expect_perm_recur, any_order=True)
+        if test_effective:
+            body = self.check_GetUserResourcesPermissions(self.usr, resource_id=test_child_res_id, query=q_effect)
+            utils.check_all_equal(body["permission_names"], expect_perm_recur, any_order=True)
         body = self.check_GetUserResourcesPermissions(self.usr, resource_id=test_parent_res_id, query=None)
         utils.check_val_equal(body["permission_names"], [])
         body = self.check_GetUserResourcesPermissions(self.usr, resource_id=test_parent_res_id, query=q_groups)
         utils.check_all_equal(body["permission_names"], expect_perm_match, any_order=True)
-        body = self.check_GetUserResourcesPermissions(self.usr, resource_id=test_parent_res_id, query=q_effect)
-        utils.check_all_equal(body["permission_names"], expect_perm_recur + expect_perm_match, any_order=True)
+        if test_effective:
+            body = self.check_GetUserResourcesPermissions(self.usr, resource_id=test_parent_res_id, query=q_effect)
+            utils.check_all_equal(body["permission_names"], expect_perm_recur + expect_perm_match, any_order=True)
         body = self.check_GetUserResourcesPermissions(self.usr, resource_id=test_svc_res_id, query=None)
         utils.check_all_equal(body["permission_names"], expect_perm_recur, any_order=True)
         body = self.check_GetUserResourcesPermissions(self.usr, resource_id=test_svc_res_id, query=q_groups)
         utils.check_all_equal(body["permission_names"], expect_perm_recur + expect_perm_match, any_order=True)
-        body = self.check_GetUserResourcesPermissions(self.usr, resource_id=test_svc_res_id, query=q_effect)
-        utils.check_all_equal(body["permission_names"], expect_perm_recur + expect_perm_match, any_order=True)
+        if test_effective:
+            body = self.check_GetUserResourcesPermissions(self.usr, resource_id=test_svc_res_id, query=q_effect)
+            utils.check_all_equal(body["permission_names"], expect_perm_recur + expect_perm_match, any_order=True)
 
         body = self.check_GetUserResourcesPermissions(anonym_usr, resource_id=test_child_res_id, query=None)
         utils.check_all_equal(body["permission_names"], expect_perm_match, any_order=True)
         body = self.check_GetUserResourcesPermissions(anonym_usr, resource_id=test_child_res_id, query=q_groups)
         utils.check_all_equal(body["permission_names"], expect_perm_match, any_order=True)
-        body = self.check_GetUserResourcesPermissions(anonym_usr, resource_id=test_child_res_id, query=q_effect)
-        utils.check_all_equal(body["permission_names"], expect_perm_recur + expect_perm_match, any_order=True)
+        if test_effective:
+            body = self.check_GetUserResourcesPermissions(anonym_usr, resource_id=test_child_res_id, query=q_effect)
+            utils.check_all_equal(body["permission_names"], expect_perm_recur + expect_perm_match, any_order=True)
         body = self.check_GetUserResourcesPermissions(anonym_usr, resource_id=test_parent_res_id, query=None)
         utils.check_val_equal(body["permission_names"], [])
         body = self.check_GetUserResourcesPermissions(anonym_usr, resource_id=test_parent_res_id, query=q_groups)
         utils.check_val_equal(body["permission_names"], [])
-        body = self.check_GetUserResourcesPermissions(anonym_usr, resource_id=test_parent_res_id, query=q_effect)
-        utils.check_all_equal(body["permission_names"], expect_perm_recur, any_order=True)
+        if test_effective:
+            body = self.check_GetUserResourcesPermissions(anonym_usr, resource_id=test_parent_res_id, query=q_effect)
+            utils.check_all_equal(body["permission_names"], expect_perm_recur, any_order=True)
         body = self.check_GetUserResourcesPermissions(anonym_usr, resource_id=test_svc_res_id, query=None)
         utils.check_val_equal(body["permission_names"], [])
         body = self.check_GetUserResourcesPermissions(anonym_usr, resource_id=test_svc_res_id, query=q_groups)
         utils.check_all_equal(body["permission_names"], expect_perm_recur, any_order=True)
-        body = self.check_GetUserResourcesPermissions(anonym_usr, resource_id=test_svc_res_id, query=q_effect)
-        utils.check_all_equal(body["permission_names"], expect_perm_recur, any_order=True)
+        if test_effective:
+            body = self.check_GetUserResourcesPermissions(anonym_usr, resource_id=test_svc_res_id, query=q_effect)
+            utils.check_all_equal(body["permission_names"], expect_perm_recur, any_order=True)
 
     @runner.MAGPIE_TEST_USERS
     @runner.MAGPIE_TEST_RESOURCES
@@ -1771,12 +1784,16 @@ def test_DeleteUserServicePermission(self):
     @runner.MAGPIE_TEST_RESOURCES
     @runner.MAGPIE_TEST_PERMISSIONS
     @runner.MAGPIE_TEST_FUNCTIONAL
-    def test_UserResourcePermissions_EffectiveResolution(self):
+    def test_GetUserResourcePermissions_EffectiveResolution(self):
         """
         Test effective resolution of permissions.
 
         Validates combinations of user/group inheritance combined with allow/deny and match/recursive modifiers.
 
+        All resolved effective permissions are scoped as :attr:`Scope.MATCH` since they target a specific resource
+        with pre-computed recursive resource tree inheritance. They should also have :attr:`PermissionType.EFFECTIVE`
+        and the corresponding :class:`Scope` value whether the permission was set or not for the resolved groups/user.
+
         Legend::
 
             r: read
@@ -2532,19 +2549,29 @@ def test_GetUserServiceResourcePermission_InheritedPublic_DirectPermission(self)
         utils.TestSetup.create_TestService(self)
         body = utils.TestSetup.create_TestServiceResource(self)
         info = utils.TestSetup.get_ResourceInfo(self, override_body=body, full_detail=True)
-        applicable_perm = info["permission_names"][0]
+        applicable_perms = info["permission_names"]
+        applied_perm = applicable_perms[0]
         res_id = info["resource_id"]
         path = "/groups/{}/resources/{}/permissions".format(get_constant("MAGPIE_ANONYMOUS_GROUP"), res_id)
-        data = {"permission_name": applicable_perm}
+        data = {"permission_name": applied_perm}
         resp = utils.test_request(self, "POST", path, json=data, headers=self.json_headers, cookies=self.cookies)
         utils.check_response_basic_info(resp, 201, expected_method="POST")
 
+        effective_perm_names = [applied_perm]
+        if LooseVersion(self.version) >= LooseVersion("2.1"):
+            effective_perm = PermissionSet(applied_perm, scope=Scope.MATCH)
+            allowed_perm_names = utils.TestSetup.get_PermissionNames(self, effective_perm)
+            denied_perm_names = {PermissionSet(perm).name for perm in applicable_perms} - {effective_perm.name}
+            denied_perms = [PermissionSet(perm, access=Access.DENY, scope=Scope.MATCH) for perm in denied_perm_names]
+            denied_perm_names = utils.TestSetup.get_PermissionNames(self, denied_perms)
+            effective_perm_names = allowed_perm_names + denied_perm_names
+
         # test
         path = "/users/{}/resources/{}/permissions?effective=true".format(self.test_user_name, res_id)
         resp = utils.test_request(self, "GET", path, headers=self.json_headers, cookies=self.cookies)
         body = utils.check_response_basic_info(resp)
         utils.check_val_is_in("permission_names", body)
-        utils.check_val_is_in(applicable_perm, body["permission_names"],
+        utils.check_all_equal(effective_perm_names, body["permission_names"], any_order=True,
                               msg="Permission applied to anonymous group which user is member of should be effective")
 
     @runner.MAGPIE_TEST_USERS
@@ -2587,10 +2614,12 @@ def test_GetUserServiceResourcePermission_InheritedPublic_ParentPermission(self)
         utils.check_val_is_in(effective_perm.implicit_permission, body["permission_names"],
                               msg="Permission applied to anonymous group which user is member of should be effective")
         if LooseVersion(self.version) >= LooseVersion("2.1"):
-            effective_permissions = [PermissionSet(perm, Access.DENY, Scope.MATCH, PermissionType.EFFECTIVE)
-                                     for perm in (set(applicable_perms) - set(applied_perm))] + [effective_perm]
+            perms_names = {PermissionSet(perm).name for perm in applicable_perms}
+            perms_denied = [PermissionSet(perm, Access.DENY, Scope.MATCH, PermissionType.EFFECTIVE)
+                            for perm in perms_names if perm != PermissionSet(applied_perm).name]
+            effective_permissions = [effective_perm] + perms_denied
             utils.check_val_is_in("permissions", body)
-            utils.check_all_equal([perm.json() for perm in effective_permissions], body["permissions"])
+            utils.check_all_equal([perm.json() for perm in effective_permissions], body["permissions"], any_order=True)
 
     @runner.MAGPIE_TEST_USERS
     def test_PostUsers(self):
@@ -3833,7 +3862,8 @@ def test_PostServiceResources_ChildrenResource_ParentID_InvalidRoot(self):
         }
         resp = utils.test_request(self, "POST", path, data=data, expect_errors=True,
                                   headers=self.json_headers, cookies=self.cookies)
-        body = utils.check_response_basic_info(resp, 400, expected_method="POST")
+        code = 403 if LooseVersion(self.version) >= LooseVersion("2.1") else 400
+        body = utils.check_response_basic_info(resp, code, expected_method="POST")
         utils.check_error_param_structure(body, version=self.version, param_compare_exists=True, param_name="parent_id")
 
     @runner.MAGPIE_TEST_SERVICES