diff --git a/lib/galaxy/managers/datasets.py b/lib/galaxy/managers/datasets.py index c7cd192593f..3d1b1ee7296 100644 --- a/lib/galaxy/managers/datasets.py +++ b/lib/galaxy/managers/datasets.py @@ -16,7 +16,6 @@ from galaxy.managers import ( base, deletable, rbac_secured, - roles, secured, users ) @@ -87,7 +86,7 @@ class DatasetManager(base.ModelManager, secured.AccessibleManagerMixin, deletabl """ Is this dataset readable/viewable to user? """ - if self.user_manager.is_admin(user): + if self.user_manager.is_admin(user, trans=kwargs.get("trans", None)): return True if self.has_access_permission(dataset, user): return True @@ -189,7 +188,7 @@ class DatasetSerializer(base.ModelSerializer, deletable.PurgableSerializerMixin) If the config allows or the user is admin, return the file name of the file that contains this dataset's data. """ - is_admin = self.user_manager.is_admin(user) + is_admin = self.user_manager.is_admin(user, trans=context.get("trans", None)) # expensive: allow config option due to cost of operation if is_admin or self.app.config.expose_dataset_path: if not dataset.purged: @@ -200,7 +199,7 @@ class DatasetSerializer(base.ModelSerializer, deletable.PurgableSerializerMixin) """ If the config allows or the user is admin, return the file path. """ - is_admin = self.user_manager.is_admin(user) + is_admin = self.user_manager.is_admin(user, trans=context.get("trans", None)) # expensive: allow config option due to cost of operation if is_admin or self.app.config.expose_dataset_path: if not dataset.purged: @@ -210,7 +209,8 @@ class DatasetSerializer(base.ModelSerializer, deletable.PurgableSerializerMixin) def serialize_permissions(self, dataset, key, user=None, **context): """ """ - if not self.dataset_manager.permissions.manage.is_permitted(dataset, user): + trans = context.get("trans", None) + if not self.dataset_manager.permissions.manage.is_permitted(dataset, user, trans=trans): self.skip() management_permissions = self.dataset_manager.permissions.manage.by_dataset(dataset) @@ -222,57 +222,6 @@ class DatasetSerializer(base.ModelSerializer, deletable.PurgableSerializerMixin) return permissions -class DatasetDeserializer(base.ModelDeserializer, deletable.PurgableDeserializerMixin): - model_manager_class = DatasetManager - - def __init__(self, app): - super(DatasetDeserializer, self).__init__(app) - # TODO: this manager may make more sense inside rbac_secured - self.role_manager = roles.RoleManager(app) - - def add_deserializers(self): - super(DatasetDeserializer, self).add_deserializers() - # not much to set here besides permissions and purged/deleted - deletable.PurgableDeserializerMixin.add_deserializers(self) - - self.deserializers.update({ - 'permissions' : self.deserialize_permissions, - }) - - def deserialize_permissions(self, dataset, key, permissions, user=None, **context): - """ - Create permissions for each list of encoded role ids in the (validated) - `permissions` dictionary, where `permissions` is in the form:: - - { 'manage': [ , ... ], 'access': [ , ... ] } - """ - self.manager.permissions.manage.error_unless_permitted(dataset, user) - self._validate_permissions(permissions, **context) - manage = self._list_of_roles_from_ids(permissions['manage']) - access = self._list_of_roles_from_ids(permissions['access']) - self.manager.permissions.set(dataset, manage, access, flush=False) - return permissions - - def _validate_permissions(self, permissions, **context): - self.validate.type('permissions', permissions, dict) - for permission_key in ('manage', 'access'): - if(not isinstance(permissions.get(permission_key, None), list)): - msg = 'permissions requires "{0}" as a list of role ids'.format(permission_key) - raise exceptions.RequestParameterInvalidException(msg) - - # TODO: push down into permissions? - manage_permissions = permissions['manage'] - if len(manage_permissions) < 1: - raise exceptions.RequestParameterInvalidException('At least one managing role is required') - - return permissions - - def _list_of_roles_from_ids(self, id_list): - # TODO: this may make more sense inside rbac_secured - # note: no checking of valid roles is made - return self.role_manager.by_ids([self.app.security.decode_id(id_) for id_ in id_list]) - - # ============================================================================= AKA DatasetInstanceManager class DatasetAssociationManager(base.ModelManager, secured.AccessibleManagerMixin, @@ -297,7 +246,7 @@ class DatasetAssociationManager(base.ModelManager, Is this DA accessible to `user`? """ # defer to the dataset - return self.dataset_manager.is_accessible(dataset_assoc.dataset, user) + return self.dataset_manager.is_accessible(dataset_assoc.dataset, user, **kwargs) def purge(self, dataset_assoc, flush=True): """ @@ -365,6 +314,85 @@ class DatasetAssociationManager(base.ModelManager, return [] return glob.glob(os.path.join(dataset_assoc.dataset.extra_files_path, '*')) + def serialize_dataset_association_roles(self, trans, dataset_assoc): + if hasattr(dataset_assoc, "library_dataset_dataset_association"): + library_dataset = dataset_assoc + dataset = library_dataset.library_dataset_dataset_association.dataset + else: + library_dataset = None + dataset = dataset_assoc.dataset + + # Omit duplicated roles by converting to set + access_roles = set(dataset.get_access_roles(trans)) + manage_roles = set(dataset.get_manage_permissions_roles(trans)) + + access_dataset_role_list = [(access_role.name, trans.security.encode_id(access_role.id)) for access_role in access_roles] + manage_dataset_role_list = [(manage_role.name, trans.security.encode_id(manage_role.id)) for manage_role in manage_roles] + rval = dict(access_dataset_roles=access_dataset_role_list, manage_dataset_roles=manage_dataset_role_list) + if library_dataset is not None: + modify_roles = set(trans.app.security_agent.get_roles_for_action(library_dataset, trans.app.security_agent.permitted_actions.LIBRARY_MODIFY)) + modify_item_role_list = [(modify_role.name, trans.security.encode_id(modify_role.id)) for modify_role in modify_roles] + rval["modify_item_roles"] = modify_item_role_list + return rval + + def update_permissions(self, trans, dataset_assoc, **kwd): + action = kwd.get('action', 'set_permissions') + if action not in ['remove_restrictions', 'make_private', 'set_permissions']: + raise exceptions.RequestParameterInvalidException('The mandatory parameter "action" has an invalid value. ' + 'Allowed values are: "remove_restrictions", "make_private", "set_permissions"') + if hasattr(dataset_assoc, "library_dataset_dataset_association"): + library_dataset = dataset_assoc + dataset = library_dataset.library_dataset_dataset_association.dataset + else: + library_dataset = None + dataset = dataset_assoc.dataset + + current_user_roles = trans.get_current_user_roles() + can_manage = trans.app.security_agent.can_manage_dataset(current_user_roles, dataset) or trans.user_is_admin() + if not can_manage: + raise exceptions.InsufficientPermissionsException('You do not have proper permissions to manage permissions on this dataset.') + + if action == 'remove_restrictions': + trans.app.security_agent.make_dataset_public(dataset) + if not trans.app.security_agent.dataset_is_public(dataset): + raise exceptions.InternalServerError('An error occured while making dataset public.') + elif action == 'make_private': + if not trans.app.security_agent.dataset_is_private_to_user(trans, dataset): + private_role = trans.app.security_agent.get_private_user_role(trans.user) + dp = trans.app.model.DatasetPermissions(trans.app.security_agent.permitted_actions.DATASET_ACCESS.action, dataset, private_role) + trans.sa_session.add(dp) + trans.sa_session.flush() + if not trans.app.security_agent.dataset_is_private_to_user(trans, dataset): + # Check again and inform the user if dataset is not private. + raise exceptions.InternalServerError('An error occured and the dataset is NOT private.') + elif action == 'set_permissions': + + def to_role_id(encoded_role_id): + role_id = base.decode_id(self.app, encoded_role_id) + return role_id + + def parameters_roles_or_none(role_type): + encoded_role_ids = kwd.get(role_type, kwd.get("%s_ids[]" % role_type, None)) + if encoded_role_ids is not None: + return list(map(to_role_id, encoded_role_ids)) + else: + return None + + access_roles = parameters_roles_or_none('access') + manage_roles = parameters_roles_or_none('manage') + modify_roles = parameters_roles_or_none('modify') + role_ids_dict = { + 'DATASET_MANAGE_PERMISSIONS': manage_roles, + 'DATASET_ACCESS': access_roles, + } + if library_dataset is not None: + role_ids_dict["LIBRARY_MODIFY"] = modify_roles + + self._set_permissions(trans, dataset_assoc, role_ids_dict) + + def _set_permissions(self, trans, dataset_assoc, roles_dict): + raise galaxy.exceptions.NotImplemented() + class _UnflattenedMetadataDatasetAssociationSerializer(base.ModelSerializer, deletable.PurgableSerializerMixin): diff --git a/lib/galaxy/managers/hdas.py b/lib/galaxy/managers/hdas.py index 29ccd342b24..0678024998b 100644 --- a/lib/galaxy/managers/hdas.py +++ b/lib/galaxy/managers/hdas.py @@ -57,14 +57,14 @@ class HDAManager(datasets.DatasetAssociationManager, # I can not access that dataset even if it's in my history # if self.is_owner( hda, user, **kwargs ): # return True - return super(HDAManager, self).is_accessible(hda, user) + return super(HDAManager, self).is_accessible(hda, user, **kwargs) def is_owner(self, hda, user, current_history=None, **kwargs): """ Use history to see if current user owns HDA. """ history = hda.history - if self.user_manager.is_admin(user): + if self.user_manager.is_admin(user, trans=kwargs.get("trans", None)): return True # allow anonymous user to access current history # TODO: some dup here with historyManager.is_owner but prevents circ import @@ -241,6 +241,23 @@ class HDAManager(datasets.DatasetAssociationManager, # override to scope to history owner return self._user_annotation(hda, hda.history.user) + def _set_permissions(self, trans, hda, role_ids_dict): + # The user associated the DATASET_ACCESS permission on the dataset with 1 or more roles. We + # need to ensure that they did not associate roles that would cause accessibility problems. + permissions, in_roles, error, message = \ + trans.app.security_agent.derive_roles_from_access(trans, hda.dataset.id, 'root', **role_ids_dict) + if error: + # Keep the original role associations for the DATASET_ACCESS permission on the dataset. + access_action = trans.app.security_agent.get_action(trans.app.security_agent.permitted_actions.DATASET_ACCESS.action) + permissions[access_action] = hda.dataset.get_access_roles(trans) + trans.sa_session.refresh(hda.dataset) + raise exceptions.RequestParameterInvalidException(message) + else: + error = trans.app.security_agent.set_all_dataset_permissions(hda.dataset, permissions) + trans.sa_session.refresh(hda.dataset) + if error: + raise exceptions.RequestParameterInvalidException(error) + class HDASerializer( # datasets._UnflattenedMetadataDatasetAssociationSerializer, datasets.DatasetAssociationSerializer, @@ -354,7 +371,7 @@ class HDASerializer( # datasets._UnflattenedMetadataDatasetAssociationSerialize history_content_id=self.app.security.encode_id(i.id)), 'parent_id' : self.serialize_id, # TODO: to DatasetAssociationSerializer - 'accessible' : lambda i, k, user=None, **c: self.manager.is_accessible(i, user), + 'accessible' : lambda i, k, user=None, **c: self.manager.is_accessible(i, user, **c), 'api_type' : lambda *a, **c: 'file', 'type' : lambda *a, **c: 'file' }) diff --git a/lib/galaxy/managers/lddas.py b/lib/galaxy/managers/lddas.py index 2aa5e384c57..d5c6beac40c 100644 --- a/lib/galaxy/managers/lddas.py +++ b/lib/galaxy/managers/lddas.py @@ -1,10 +1,17 @@ +import logging + +from galaxy import model, util from galaxy.managers import base as manager_base +from galaxy.managers.datasets import DatasetAssociationManager + +log = logging.getLogger(__name__) -class LDDAManager(object): +class LDDAManager(DatasetAssociationManager): """ A fairly sparse manager for LDDAs. """ + model_class = model.LibraryDatasetDatasetAssociation def __init__(self, app): """ @@ -17,3 +24,58 @@ class LDDAManager(object): 'LibraryDatasetDatasetAssociation', check_ownership=False, check_accessible=check_accessible) + + def _set_permissions(self, trans, library_dataset, role_ids_dict): + dataset = library_dataset.library_dataset_dataset_association.dataset + new_access_roles_ids = role_ids_dict["DATASET_ACCESS"] + new_manage_roles_ids = role_ids_dict["DATASET_MANAGE_PERMISSIONS"] + new_modify_roles_ids = role_ids_dict["LIBRARY_MODIFY"] + + # ACCESS DATASET ROLES + valid_access_roles = [] + invalid_access_roles_ids = [] + valid_roles_for_dataset, total_roles = trans.app.security_agent.get_valid_roles(trans, dataset) + if new_access_roles_ids is None: + trans.app.security_agent.make_dataset_public(dataset) + else: + for role_id in new_access_roles_ids: + role = self.role_manager.get(trans, self.app, role_id) + if role in valid_roles_for_dataset: + valid_access_roles.append(role) + else: + invalid_access_roles_ids.append(role_id) + if len(invalid_access_roles_ids) > 0: + log.warning("The following roles could not be added to the dataset access permission: " + str(invalid_access_roles_ids)) + + access_permission = dict(access=valid_access_roles) + trans.app.security_agent.set_dataset_permission(dataset, access_permission) + + # MANAGE DATASET ROLES + valid_manage_roles = [] + invalid_manage_roles_ids = [] + new_manage_roles_ids = util.listify(new_manage_roles_ids) + for role_id in new_manage_roles_ids: + role = self.role_manager.get(trans, self.app, role_id) + if role in valid_roles_for_dataset: + valid_manage_roles.append(role) + else: + invalid_manage_roles_ids.append(role_id) + if len(invalid_manage_roles_ids) > 0: + log.warning("The following roles could not be added to the dataset manage permission: " + str(invalid_manage_roles_ids)) + manage_permission = {trans.app.security_agent.permitted_actions.DATASET_MANAGE_PERMISSIONS: valid_manage_roles} + trans.app.security_agent.set_dataset_permission(dataset, manage_permission) + + # MODIFY LIBRARY ITEM ROLES + valid_modify_roles = [] + invalid_modify_roles_ids = [] + new_modify_roles_ids = util.listify(new_modify_roles_ids) + for role_id in new_modify_roles_ids: + role = self.role_manager.get(trans, self.app, role_id) + if role in valid_roles_for_dataset: + valid_modify_roles.append(role) + else: + invalid_modify_roles_ids.append(role_id) + if len(invalid_modify_roles_ids) > 0: + log.warning("The following roles could not be added to the dataset modify permission: " + str(invalid_modify_roles_ids)) + modify_permission = {trans.app.security_agent.permitted_actions.LIBRARY_MODIFY: valid_modify_roles} + trans.app.security_agent.set_library_item_permission(library_dataset, modify_permission) diff --git a/lib/galaxy/managers/rbac_secured.py b/lib/galaxy/managers/rbac_secured.py index 0f473008092..262b46f8897 100644 --- a/lib/galaxy/managers/rbac_secured.py +++ b/lib/galaxy/managers/rbac_secured.py @@ -33,11 +33,11 @@ class RBACPermission(object): # TODO: implement group # TODO: how does admin play into this? - def is_permitted(self, item, user): + def is_permitted(self, item, user, trans=None): raise NotImplementedError("abstract parent class") - def error_unless_permitted(self, item, user): - if not self.is_permitted(item, user): + def error_unless_permitted(self, item, user, trans=None): + if not self.is_permitted(item, user, trans=trans): error_info = dict(model_class=item.__class__, id=getattr(item, 'id', None)) raise self.permission_failed_error_class(**error_info) @@ -172,7 +172,10 @@ class ManageDatasetRBACPermission(DatasetRBACPermission): permission_failed_error_class = DatasetManagePermissionFailedException # ---- interface - def is_permitted(self, dataset, user): + def is_permitted(self, dataset, user, trans=None): + if trans and trans.user_is_admin(): + return True + # anonymous users cannot manage permissions on datasets if self.user_manager.is_anonymous(user): return False @@ -228,7 +231,10 @@ class AccessDatasetRBACPermission(DatasetRBACPermission): permission_failed_error_class = DatasetAccessPermissionFailedException # ---- interface - def is_permitted(self, dataset, user): + def is_permitted(self, dataset, user, trans=None): + if trans and trans.user_is_admin(): + return True + current_roles = self._roles(dataset) # NOTE: that because of short circuiting this allows # anonymous access to public datasets diff --git a/lib/galaxy/managers/sharable.py b/lib/galaxy/managers/sharable.py index 2c596a66458..790fc79ea7e 100644 --- a/lib/galaxy/managers/sharable.py +++ b/lib/galaxy/managers/sharable.py @@ -59,7 +59,7 @@ class SharableModelManager(base.ModelManager, secured.OwnableManagerMixin, secur Return true if this sharable belongs to `user` (or `user` is an admin). """ # ... effectively a good fit to have this here, but not semantically - if self.user_manager.is_admin(user): + if self.user_manager.is_admin(user, trans=kwargs.get("trans", None)): return True return item.user == user diff --git a/lib/galaxy/managers/users.py b/lib/galaxy/managers/users.py index 97fb4d5c00c..45cf1d4d910 100644 --- a/lib/galaxy/managers/users.py +++ b/lib/galaxy/managers/users.py @@ -103,12 +103,20 @@ class UserManager(base.ModelManager, deletable.PurgableManagerMixin): return super(UserManager, self).list(filters=filters, order_by=order_by, **kwargs) # ---- admin - def is_admin(self, user): - """ - Return True if this user is an admin. + def is_admin(self, user, trans=None): + """Return True if this user is an admin (or session is authenticated as admin). + + Do not pass trans to simply check if an existing user object is an admin user, + pass trans when checking permissions. """ admin_emails = self._admin_emails() - return user and admin_emails and user.email in admin_emails + if user is None: + # Anonymous session or master_api_key used, if master_api_key is detected + # return True. + log.info("in is_admin.... %s" % trans) + rval = bool(trans and trans.user_is_admin()) + return rval + return bool(admin_emails and user.email in admin_emails) def _admin_emails(self): """ @@ -130,7 +138,7 @@ class UserManager(base.ModelManager, deletable.PurgableManagerMixin): :raises exceptions.AdminRequiredException: if `user` is not an admin. """ # useful in admin only methods - if not self.is_admin(user): + if not self.is_admin(user, trans=kwargs.get("trans", None)): raise exceptions.AdminRequiredException(msg, **kwargs) return user diff --git a/lib/galaxy/security/__init__.py b/lib/galaxy/security/__init__.py index 654c5d85b58..3d3fe09c208 100644 --- a/lib/galaxy/security/__init__.py +++ b/lib/galaxy/security/__init__.py @@ -1129,11 +1129,11 @@ class GalaxyRBACAgent(RBACAgent): def dataset_is_private_to_user(self, trans, dataset): """ - If the LibraryDataset object has exactly one access role and that is + If the Dataset object has exactly one access role and that is the current user's private role then we consider the dataset private. """ private_role = self.get_private_user_role(trans.user) - access_roles = dataset.library_dataset_dataset_association.get_access_roles(trans) + access_roles = dataset.get_access_roles(trans) if len(access_roles) != 1: return False diff --git a/lib/galaxy/web/base/controller.py b/lib/galaxy/web/base/controller.py index 12866f2a7bd..14bf58dda18 100644 --- a/lib/galaxy/web/base/controller.py +++ b/lib/galaxy/web/base/controller.py @@ -292,7 +292,7 @@ class JSAppLauncher(BaseUIController): """ try: serializer = self.config_serializer - if self.user_manager.is_admin(trans.user): + if self.user_manager.is_admin(trans.user, trans=trans): serializer = self.admin_config_serializer return serializer.serialize_to_view(self.app.config, view='all') except Exception as exc: diff --git a/lib/galaxy/webapps/galaxy/api/datasets.py b/lib/galaxy/webapps/galaxy/api/datasets.py index d63a97f4f7a..63d5d006d04 100644 --- a/lib/galaxy/webapps/galaxy/api/datasets.py +++ b/lib/galaxy/webapps/galaxy/api/datasets.py @@ -33,6 +33,7 @@ class DatasetsController(BaseAPIController, UsesVisualizationMixin): super(DatasetsController, self).__init__(app) self.hda_manager = managers.hdas.HDAManager(app) self.hda_serializer = managers.hdas.HDASerializer(self.app) + self.ldda_manager = managers.lddas.LDDAManager(app) def _parse_serialization_params(self, kwd, default_view): view = kwd.get('view', None) @@ -93,6 +94,26 @@ class DatasetsController(BaseAPIController, UsesVisualizationMixin): trans.response.status = 500 return rval + @web._future_expose_api + def update_permissions(self, trans, dataset_id, payload, **kwd): + """ + PUT /api/datasets/{encoded_dataset_id}/permissions + Updates permissions of a dataset. + + :rtype: dict + :returns: dictionary containing new permissions + """ + if payload: + kwd.update(payload) + hda_ldda = kwd.get('hda_ldda', 'hda') + dataset_assoc = self.get_hda_or_ldda(trans, hda_ldda=hda_ldda, dataset_id=dataset_id) + if hda_ldda == "hda": + self.hda_manager.update_permissions(trans, dataset_assoc, **kwd) + return self.hda_manager.serialize_dataset_association_roles(trans, dataset_assoc) + else: + self.ldda_manager.update_permissions(trans, dataset_assoc, **kwd) + return self.hda_manager.serialize_dataset_association_roles(trans, dataset_assoc) + def _dataset_state(self, trans, dataset, **kwargs): """ Returns state of dataset. diff --git a/lib/galaxy/webapps/galaxy/api/folder_contents.py b/lib/galaxy/webapps/galaxy/api/folder_contents.py index b012d4a4c59..8d4f1b9634c 100644 --- a/lib/galaxy/webapps/galaxy/api/folder_contents.py +++ b/lib/galaxy/webapps/galaxy/api/folder_contents.py @@ -96,8 +96,9 @@ class FolderContentsController(BaseAPIController, UsesLibraryMixin, UsesLibraryM # Is the dataset public or private? # When both are False the dataset is 'restricted' # Access rights are checked on the dataset level, not on the ld or ldda level to maintain consistency - is_unrestricted = trans.app.security_agent.dataset_is_public(content_item.library_dataset_dataset_association.dataset) - if trans.user and trans.app.security_agent.dataset_is_private_to_user(trans, content_item): + dataset = content_item.library_dataset_dataset_association.dataset + is_unrestricted = trans.app.security_agent.dataset_is_public(dataset) + if trans.user and trans.app.security_agent.dataset_is_private_to_user(trans, dataset): is_private = True else: is_private = False diff --git a/lib/galaxy/webapps/galaxy/api/history_contents.py b/lib/galaxy/webapps/galaxy/api/history_contents.py index 07706d2e979..e010b8b1d00 100644 --- a/lib/galaxy/webapps/galaxy/api/history_contents.py +++ b/lib/galaxy/webapps/galaxy/api/history_contents.py @@ -553,6 +553,59 @@ class HistoryContentsController(BaseAPIController, UsesLibraryMixin, UsesLibrary return self.__collection_dict(trans, dataset_collection_instance, view="element") + @expose_api + def show_roles(self, trans, encoded_dataset_id, **kwd): + """ + Display information about current or available roles for a given dataset permission. + + * GET /api/histories/{history_id}/contents/datasets/{encoded_dataset_id}/permissions + + :param encoded_dataset_id: the encoded id of the dataset to query + :type encoded_dataset_id: an encoded id string + + :returns: either dict of current roles for all permission types + or dict of available roles to choose from (is the same for any permission type) + :rtype: dictionary + + :raises: InsufficientPermissionsException + """ + hda = self.hda_manager.get_owned(self.decode_id(encoded_dataset_id), trans.user, current_history=trans.history, trans=trans) + return self.hda_manager.serialize_dataset_association_roles(trans, hda) + + @expose_api + def update_permissions(self, trans, history_id, history_content_id, payload=None, **kwd): + """ + Set permissions of the given library dataset to the given role ids. + + * PUT /api/histories/{history_id}/contents/datasets/{encoded_dataset_id}/permissions + + :param encoded_dataset_id: the encoded id of the dataset to update permissions of + :type encoded_dataset_id: an encoded id string + :param payload: dictionary structure containing: + :param action: (required) describes what action should be performed + available actions: make_private, remove_restrictions, set_permissions + :type action: string + :param access_ids[]: list of Role.id defining roles that should have access permission on the dataset + :type access_ids[]: string or list + :param manage_ids[]: list of Role.id defining roles that should have manage permission on the dataset + :type manage_ids[]: string or list + :param modify_ids[]: list of Role.id defining roles that should have modify permission on the library dataset item + :type modify_ids[]: string or list + :type: dictionary + + :returns: dict of current roles for all available permission types + :rtype: dictionary + + :raises: RequestParameterInvalidException, ObjectNotFound, InsufficientPermissionsException, InternalServerError + RequestParameterMissingException + """ + if payload: + kwd.update(payload) + hda = self.hda_manager.get_owned(self.decode_id(history_content_id), trans.user, current_history=trans.history, trans=trans) + assert hda is not None + self.hda_manager.update_permissions(trans, hda, **kwd) + return self.hda_manager.serialize_dataset_association_roles(trans, hda) + @expose_api_anonymous def update(self, trans, history_id, id, payload, **kwd): """ @@ -587,7 +640,8 @@ class HistoryContentsController(BaseAPIController, UsesLibraryMixin, UsesLibrary def __update_dataset(self, trans, history_id, id, payload, **kwd): # anon user: ensure that history ids match up and the history is the current, # check for uploading, and use only the subset of attribute keys manipulatable by anon users - if trans.user is None: + anonymous_user = not trans.user_is_admin() and trans.user is None + if anonymous_user: hda = self.hda_manager.by_id(self.decode_id(id)) if hda.history != trans.history: raise exceptions.AuthenticationRequired('API authentication required for this request') @@ -602,7 +656,7 @@ class HistoryContentsController(BaseAPIController, UsesLibraryMixin, UsesLibrary # logged in user: use full payload, check state if deleting, and make sure the history is theirs else: - hda = self.hda_manager.get_owned(self.decode_id(id), trans.user, current_history=trans.history) + hda = self.hda_manager.get_owned(self.decode_id(id), trans.user, current_history=trans.history, trans=trans) # only check_state if not deleting, otherwise cannot delete uploading files check_state = not payload.get('deleted', False) diff --git a/lib/galaxy/webapps/galaxy/api/library_datasets.py b/lib/galaxy/webapps/galaxy/api/library_datasets.py index e41ad7ea81b..7c5b9fd3da1 100644 --- a/lib/galaxy/webapps/galaxy/api/library_datasets.py +++ b/lib/galaxy/webapps/galaxy/api/library_datasets.py @@ -20,6 +20,7 @@ from galaxy.exceptions import ObjectNotFound from galaxy.managers import ( base as managers_base, folders, + lddas, library_datasets, roles ) @@ -43,6 +44,7 @@ class LibraryDatasetsController(BaseAPIController, UsesVisualizationMixin, Libra self.folder_manager = folders.FolderManager() self.role_manager = roles.RoleManager(app) self.ld_manager = library_datasets.LibraryDatasetsManager(app) + self.ldda_manager = lddas.LDDAManager(app) @expose_api_anonymous def show(self, trans, id, **kwd): @@ -153,18 +155,7 @@ class LibraryDatasetsController(BaseAPIController, UsesVisualizationMixin, Libra :rtype: dictionary :returns: dict of current roles for all available permission types """ - dataset = library_dataset.library_dataset_dataset_association.dataset - - # Omit duplicated roles by converting to set - access_roles = set(dataset.get_access_roles(trans)) - modify_roles = set(trans.app.security_agent.get_roles_for_action(library_dataset, trans.app.security_agent.permitted_actions.LIBRARY_MODIFY)) - manage_roles = set(dataset.get_manage_permissions_roles(trans)) - - access_dataset_role_list = [(access_role.name, trans.security.encode_id(access_role.id)) for access_role in access_roles] - manage_dataset_role_list = [(manage_role.name, trans.security.encode_id(manage_role.id)) for manage_role in manage_roles] - modify_item_role_list = [(modify_role.name, trans.security.encode_id(modify_role.id)) for modify_role in modify_roles] - - return dict(access_dataset_roles=access_dataset_role_list, modify_item_roles=modify_item_role_list, manage_dataset_roles=manage_dataset_role_list) + return self.serialize_dataset_association_roles(library_dataset) @expose_api def update(self, trans, encoded_dataset_id, payload=None, **kwd): @@ -241,12 +232,12 @@ class LibraryDatasetsController(BaseAPIController, UsesVisualizationMixin, Libra if not trans.app.security_agent.dataset_is_public(dataset): raise exceptions.InternalServerError('An error occured while making dataset public.') elif action == 'make_private': - if not trans.app.security_agent.dataset_is_private_to_user(trans, library_dataset): + if not trans.app.security_agent.dataset_is_private_to_user(trans, dataset): private_role = trans.app.security_agent.get_private_user_role(trans.user) dp = trans.app.model.DatasetPermissions(trans.app.security_agent.permitted_actions.DATASET_ACCESS.action, dataset, private_role) trans.sa_session.add(dp) trans.sa_session.flush() - if not trans.app.security_agent.dataset_is_private_to_user(trans, library_dataset): + if not trans.app.security_agent.dataset_is_private_to_user(trans, dataset): # Check again and inform the user if dataset is not private. raise exceptions.InternalServerError('An error occured and the dataset is NOT private.') elif action == 'set_permissions': diff --git a/lib/galaxy/webapps/galaxy/buildapp.py b/lib/galaxy/webapps/galaxy/buildapp.py index 73426d96a7f..f12d7d5d5fd 100644 --- a/lib/galaxy/webapps/galaxy/buildapp.py +++ b/lib/galaxy/webapps/galaxy/buildapp.py @@ -218,6 +218,11 @@ def populate_api_routes(webapp, app): controller="datasets", action="display", conditions=dict(method=["GET"])) + webapp.mapper.connect("history_contents_update_permissions", + "/api/histories/{history_id}/contents/{history_content_id}/permissions", + controller="history_contents", + action="update_permissions", + conditions=dict(method=["PUT"])) webapp.mapper.connect("history_contents_metadata_file", "/api/histories/{history_id}/contents/{history_content_id}/metadata_file", controller="datasets", @@ -457,6 +462,7 @@ def populate_api_routes(webapp, app): # route for creating/getting converted datasets webapp.mapper.connect('/api/datasets/{dataset_id}/converted', controller='datasets', action='converted', ext=None) webapp.mapper.connect('/api/datasets/{dataset_id}/converted/{ext}', controller='datasets', action='converted') + webapp.mapper.connect('/api/datasets/{dataset_id}/permissions', controller='datasets', action='update_permissions', conditions=dict(method=["PUT"])) # API refers to usages and invocations - these mean the same thing but the # usage routes should be considered deprecated. @@ -677,11 +683,12 @@ def populate_api_routes(webapp, app): action='get_permissions', conditions=dict(method=["GET"])) + # POST for legacy reasons, but this should be a PUT. webapp.mapper.connect('set_library_permissions', '/api/libraries/{encoded_library_id}/permissions', controller='libraries', action='set_permissions', - conditions=dict(method=["POST"])) + conditions=dict(method=["POST", "PUT"])) webapp.mapper.connect('show_ld_item', '/api/libraries/datasets/{id}', diff --git a/test/api/test_dataset_collections.py b/test/api/test_dataset_collections.py index df22a2635ec..d01a1d3c4e9 100644 --- a/test/api/test_dataset_collections.py +++ b/test/api/test_dataset_collections.py @@ -160,7 +160,7 @@ class DatasetCollectionApiTestCase(api.ApiTestCase): def test_hda_security(self): element_identifiers = self.dataset_collection_populator.pair_identifiers(self.history_id) - + self.dataset_populator.make_private(self.history_id, element_identifiers[0]["id"]) with self._different_user(): history_id = self.dataset_populator.new_history() payload = dict( @@ -169,11 +169,8 @@ class DatasetCollectionApiTestCase(api.ApiTestCase): element_identifiers=json.dumps(element_identifiers), collection_type="paired", ) - - self._post("dataset_collections", payload) - # TODO: re-enable once there is a way to restrict access - # to this dataset via the API. - # self._assert_status_code_is( create_response, 403 ) + create_response = self._post("dataset_collections", payload) + self._assert_status_code_is(create_response, 403) def test_enforces_unique_names(self): element_identifiers = self.dataset_collection_populator.list_identifiers(self.history_id) diff --git a/test/api/test_history_contents.py b/test/api/test_history_contents.py index a64278a989d..1184a17c156 100644 --- a/test/api/test_history_contents.py +++ b/test/api/test_history_contents.py @@ -30,6 +30,114 @@ class HistoryContentsApiTestCase(api.ApiTestCase, TestsDatasets): hda_summary = self.__check_for_hda(contents_response, hda1) assert "display_types" not in hda_summary # Quick summary, not full details + def test_make_private_and_public(self): + hda1 = self._wait_for_new_hda() + update_url = "histories/%s/contents/%s/permissions" % (self.history_id, hda1["id"]) + + role_id = self.dataset_populator.user_private_role_id() + # Give manage permission to the user. + payload = { + "access": [], + "manage": [role_id], + } + update_response = self._update_permissions(update_url, payload, admin=True) + self._assert_status_code_is(update_response, 200) + self._assert_other_user_can_access(hda1["id"]) + # Then we restrict access. + payload = { + "action": "make_private", + } + update_response = self._update_permissions(update_url, payload) + self._assert_status_code_is(update_response, 200) + self._assert_other_user_cannot_access(hda1["id"]) + + # Then we restrict access. + payload = { + "action": "remove_restrictions", + } + update_response = self._update_permissions(update_url, payload) + self._assert_status_code_is(update_response, 200) + self._assert_other_user_can_access(hda1["id"]) + + def test_set_permissions_add_admin_history_contents(self): + self._verify_dataset_permissions("history_contents") + + def test_set_permissions_add_admin_datasets(self): + self._verify_dataset_permissions("dataset") + + def _verify_dataset_permissions(self, api_endpoint): + hda1 = self._wait_for_new_hda() + hda_id = hda1["id"] + if api_endpoint == "history_contents": + update_url = "histories/%s/contents/%s/permissions" % (self.history_id, hda_id) + else: + update_url = "datasets/%s/permissions" % hda_id + + role_id = self.dataset_populator.user_private_role_id() + + payload = { + "access": [role_id], + "manage": [role_id], + } + + # Other users cannot modify permissions. + with self._different_user(): + update_response = self._update_permissions(update_url, payload) + self._assert_status_code_is(update_response, 403) + + # First the details render for another user. + self._assert_other_user_can_access(hda_id) + + # Then we restrict access. + update_response = self._update_permissions(update_url, payload, admin=True) + self._assert_status_code_is(update_response, 200) + + # Finally the details don't render. + self._assert_other_user_cannot_access(hda_id) + + # But they do for the original user. + contents_response = self._get("histories/%s/contents/%s" % (self.history_id, hda_id)).json() + assert "name" in contents_response + + update_response = self._update_permissions(update_url, payload) + self._assert_status_code_is(update_response, 200) + + payload = { + "access": [role_id], + "manage": [role_id], + } + update_response = self._update_permissions(update_url, payload) + self._assert_status_code_is(update_response, 200) + self._assert_other_user_cannot_access(hda_id) + + user_id = self.dataset_populator.user_id() + with self._different_user(): + different_user_id = self.dataset_populator.user_id() + combined_user_role = self.dataset_populator.create_role([user_id, different_user_id], description="role for testing permissions") + + payload = { + "access": [combined_user_role["id"]], + "manage": [role_id], + } + update_response = self._update_permissions(update_url, payload) + self._assert_status_code_is(update_response, 200) + # Now other user can see dataset again with access permission. + self._assert_other_user_can_access(hda_id) + # access doesn't imply management though... + with self._different_user(): + update_response = self._update_permissions(update_url, payload) + self._assert_status_code_is(update_response, 403) + + def _assert_other_user_cannot_access(self, history_content_id): + with self._different_user(): + contents_response = self._get("histories/%s/contents/%s" % (self.history_id, history_content_id)).json() + assert "name" not in contents_response + + def _assert_other_user_can_access(self, history_content_id): + with self._different_user(): + contents_response = self._get("histories/%s/contents/%s" % (self.history_id, history_content_id)).json() + assert "name" in contents_response + def test_index_hda_all_details(self): hda1 = self._new_dataset(self.history_id) contents_response = self._get("histories/%s/contents?details=all" % self.history_id) @@ -105,8 +213,15 @@ class HistoryContentsApiTestCase(api.ApiTestCase, TestsDatasets): self._wait_for_history(self.history_id) return hda1 - def _raw_update(self, item_id, data): - update_url = self._api_url("histories/%s/contents/%s" % (self.history_id, item_id), use_key=True) + def _raw_update(self, item_id, data, admin=False): + key_param = "use_admin_key" if admin else "use_key" + update_url = self._api_url("histories/%s/contents/%s" % (self.history_id, item_id), **{key_param: True}) + update_response = put(update_url, json=data) + return update_response + + def _update_permissions(self, url, data, admin=False): + key_param = "use_admin_key" if admin else "use_key" + update_url = self._api_url(url, **{key_param: True}) update_response = put(update_url, json=data) return update_response diff --git a/test/api/test_roles.py b/test/api/test_roles.py new file mode 100644 index 00000000000..a0d9aa6dacb --- /dev/null +++ b/test/api/test_roles.py @@ -0,0 +1,104 @@ +import json + +from base import api +from base.api_asserts import ( + assert_has_keys, + assert_status_code_is, +) +from base.populators import ( # noqa: I100 + DatasetPopulator, +) + + +class RolesApiTestCase(api.ApiTestCase): + + def setUp(self): + super(RolesApiTestCase, self).setUp() + self.dataset_populator = DatasetPopulator(self.galaxy_interactor) + + def test_list_and_show(self): + + def check_roles_response(response): + assert response.status_code == 200 + as_list = response.json() + assert isinstance(as_list, list) + assert len(as_list) > 0 + for role in as_list: + RolesApiTestCase.check_role_dict(role) + + user_role_id = self.dataset_populator.user_private_role_id() + with self._different_user(): + different_user_role_id = self.dataset_populator.user_private_role_id() + + admin_roles_response = self._get("roles", admin=True) + user_roles_response = self._get("roles") + + check_roles_response(admin_roles_response) + check_roles_response(user_roles_response) + + admin_roles_response_ids = [r["id"] for r in admin_roles_response.json()] + user_roles_response_ids = [r["id"] for r in user_roles_response.json()] + + # User can see their own private role not the other users, admin can see both. + assert user_role_id in user_roles_response_ids + assert different_user_role_id not in user_roles_response_ids + + assert user_role_id in admin_roles_response_ids + assert different_user_role_id in admin_roles_response_ids + + # Check showing a valid, role. + role_response = self._get("roles/%s" % user_role_id) + assert role_response.status_code == 200 + role = role_response.json() + RolesApiTestCase.check_role_dict(role, assert_id=user_role_id) + + def test_create_valid(self): + name = self.dataset_populator.get_random_name() + description = "A test role." + payload = { + "name": name, + "description": description, + "user_ids": json.dumps([self.dataset_populator.user_id()]), + } + response = self._post("roles", payload, admin=True) + assert_status_code_is(response, 200) + # TODO: Why does this return a singleton list - that is bad - should be deprecated + # and return a single role. + role = response.json()[0] + RolesApiTestCase.check_role_dict(role) + + assert role["name"] == name + assert role["description"] == description + + user_roles_response = self._get("roles") + with self._different_user(): + different_user_roles_response = self._get("roles") + + user_roles_response_ids = [r["id"] for r in user_roles_response.json()] + different_user_roles_response_ids = [r["id"] for r in different_user_roles_response.json()] + + # This new role is public, all users see it. + assert role["id"] in user_roles_response_ids + assert role["id"] in different_user_roles_response_ids + + def test_show_error_codes(self): + # Bad role ids are 400. + response = self._get("roles/badroleid") + assert_status_code_is(response, 400) + + # Trying to access roles are errors - should probably be 403 not 400 though? + with self._different_user(): + different_user_role_id = self.dataset_populator.user_private_role_id() + response = self._get("roles/%s" % different_user_role_id) + assert_status_code_is(response, 400) + + def test_create_only_admin(self): + response = self._post("roles") + assert_status_code_is(response, 403) + + @staticmethod + def check_role_dict(role_dict, assert_id=None): + assert_has_keys(role_dict, "id", "name", "model_class", "url") + assert role_dict["model_class"] == "Role" + if assert_id is not None: + assert role_dict["id"] == assert_id diff --git a/test/api/test_tools.py b/test/api/test_tools.py index 361f8d0d4e9..cb80c90ec21 100644 --- a/test/api/test_tools.py +++ b/test/api/test_tools.py @@ -1,4 +1,5 @@ # Test tools API. +import contextlib import json import os @@ -251,6 +252,37 @@ class ToolsTestCase(api.ApiTestCase): zipped_hdca = self.dataset_populator.get_history_collection_details(history_id, hid=output_collections[0]["hid"]) assert zipped_hdca["collection_type"] == "paired" + @skip_without_tool("__ZIP_COLLECTION__") + @uses_test_history(require_new=False) + def test_collection_operation_dataset_input_permissions(self, history_id): + hda1 = dataset_to_param(self.dataset_populator.new_dataset(history_id, content='1\t2\t3')) + self.dataset_populator.wait_for_history(history_id, assert_ok=True) + self.dataset_populator.make_private(history_id, hda1["id"]) + inputs = { + "input_forward": hda1, + "input_reverse": hda1, + } + with self._different_user_and_history() as other_history_id: + response = self._run("__ZIP_COLLECTION__", other_history_id, inputs, assert_ok=False) + self._assert_dataset_permission_denied_response(response) + + @skip_without_tool("__UNZIP_COLLECTION__") + @uses_test_history(require_new=False) + def test_collection_operation_collection_input_permissions(self, history_id): + create_response = self.dataset_collection_populator.create_pair_in_history(history_id, direct_upload=True) + self._assert_status_code_is(create_response, 200) + collection = create_response.json()["outputs"][0] + self.dataset_populator.wait_for_history(history_id, assert_ok=True) + collection = self.dataset_populator.get_history_collection_details(history_id, hid=collection["hid"]) + element_id = collection["elements"][0]["object"]["id"] + self.dataset_populator.make_private(history_id, element_id) + inputs = { + "input": {"src": "hdca", "id": collection["id"]}, + } + with self._different_user_and_history() as other_history_id: + response = self._run("__UNZIP_COLLECTION__", other_history_id, inputs, assert_ok=False) + self._assert_dataset_permission_denied_response(response) + def test_zip_list_inputs(self): with self.dataset_populator.test_history() as history_id: hdca1_id = self.dataset_collection_populator.create_list_in_history(history_id, contents=["a\nb\nc\nd", "e\nf\ng\nh"]).json()["id"] @@ -496,6 +528,79 @@ class ToolsTestCase(api.ApiTestCase): output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1) self.assertEqual(output1_content.strip(), "123") + @skip_without_tool("cat1") + @uses_test_history(require_new=False) + def test_guess_derived_permissions(self, history_id): + + def assert_inputs(inputs, can_be_used=True): + # Until we make the dataset private, _different_user() can use it: + with self._different_user_and_history() as other_history_id: + response = self._run("cat1", other_history_id, inputs) + if can_be_used: + assert response.status_code == 200 + else: + self._assert_dataset_permission_denied_response(response) + + new_dataset = self.dataset_populator.new_dataset(history_id, content='Cat1Test') + inputs = dict( + input1=dataset_to_param(new_dataset), + ) + # Until we make the dataset private, _different_user() can use it: + assert_inputs(inputs, can_be_used=True) + self.dataset_populator.make_private(history_id, new_dataset["id"]) + # _different_user can no longer use the input dataset. + assert_inputs(inputs, can_be_used=False) + + outputs = self._cat1_outputs(history_id, inputs=inputs) + self.assertEquals(len(outputs), 1) + output1 = outputs[0] + + inputs_2 = dict( + input1=dataset_to_param(output1), + ) + # _different_user cannot use datasets derived from the private input. + assert_inputs(inputs_2, can_be_used=False) + + @skip_without_tool("collection_creates_list") + @uses_test_history(require_new=False) + def test_guess_derived_permissions_collections(self, history_id): + def first_element_dataset_id(hdca): + # Fetch full and updated details for HDCA + print(hdca) + full_hdca = self.dataset_populator.get_history_collection_details(history_id, hid=hdca["hid"]) + elements = full_hdca["elements"] + element0 = elements[0]["object"] + return element0["id"] + + response = self.dataset_collection_populator.create_list_in_history(history_id, contents=["a\nb\nc\nd", "e\nf\ng\nh"], direct_upload=True) + self._assert_status_code_is(response, 200) + hdca = response.json()["output_collections"][0] + self.dataset_populator.wait_for_history(history_id, assert_ok=True) + inputs = { + "input1": {"src": "hdca", "id": hdca["id"]}, + } + public_output_response = self._run("collection_creates_list", history_id, inputs) + self._assert_status_code_is(public_output_response, 200) + self.dataset_populator.wait_for_history(history_id, assert_ok=True) + + input_element_id = first_element_dataset_id(hdca) + self.dataset_populator.make_private(history_id, input_element_id) + + private_output_response = self._run("collection_creates_list", history_id, inputs) + self._assert_status_code_is(private_output_response, 200) + self.dataset_populator.wait_for_history(history_id, assert_ok=True) + + public_element_id = first_element_dataset_id(public_output_response.json()["output_collections"][0]) + private_element_id = first_element_dataset_id(private_output_response.json()["output_collections"][0]) + + def _dataset_accessible(dataset_id): + contents_response = self._get("histories/%s/contents/%s" % (history_id, dataset_id)).json() + return "name" in contents_response + + with self._different_user(): + assert _dataset_accessible(public_element_id) + assert not _dataset_accessible(private_element_id) + @skip_without_tool("validation_default") @uses_test_history(require_new=False) def test_validation(self, history_id): @@ -1808,6 +1913,19 @@ class ToolsTestCase(api.ApiTestCase): hdca_id = create_response.json()["outputs"][0]["id"] return hdca_id + def _assert_dataset_permission_denied_response(self, response): + # TODO: This should be 403, should just need to throw more specific exception in the + # Galaxy code. + assert response.status_code != 200 + # err_message = response.json()["err_msg"] + # assert "User does not have permission to use a dataset" in err_message, err_message + + @contextlib.contextmanager + def _different_user_and_history(self): + with self._different_user(): + with self.dataset_populator.test_history() as other_history_id: + yield other_history_id + def dataset_to_param(dataset): return dict( diff --git a/test/base/api.py b/test/base/api.py index ed522e10a23..57b078ec83d 100644 --- a/test/base/api.py +++ b/test/base/api.py @@ -21,12 +21,14 @@ from .testcase import FunctionalTestCase class UsesApiTestCaseMixin: - def _api_url(self, path, params=None, use_key=None): + def _api_url(self, path, params=None, use_key=None, use_admin_key=None): if not params: params = {} url = "%s/api/%s" % (self.url, path) if use_key: params["key"] = self.galaxy_interactor.api_key + if use_admin_key: + params["key"] = self.galaxy_interactor.master_api_key query = urlencode(params) if query: url = "%s?%s" % (url, query) diff --git a/test/base/populators.py b/test/base/populators.py index 4084d410e34..58b67c375d3 100644 --- a/test/base/populators.py +++ b/test/base/populators.py @@ -1,6 +1,8 @@ import contextlib import json import os +import random +import string import time import unittest from functools import wraps @@ -422,6 +424,60 @@ class BaseDatasetPopulator(object): src = 'hdca' return dict(src=src, id=history_content["id"]) + def get_roles(self): + roles_response = self.galaxy_interactor.get("roles", admin=True) + assert roles_response.status_code == 200 + return roles_response.json() + + def user_email(self): + users_response = self.galaxy_interactor.get("users") + users = users_response.json() + assert len(users) == 1 + return users[0]["email"] + + def user_id(self): + users_response = self.galaxy_interactor.get("users") + users = users_response.json() + assert len(users) == 1 + return users[0]["id"] + + def user_private_role_id(self): + user_email = self.user_email() + roles = self.get_roles() + users_roles = [r for r in roles if r["name"] == user_email] + assert len(users_roles) == 1 + return users_roles[0]["id"] + + def create_role(self, user_ids, description=None): + payload = { + "name": self.get_random_name(prefix="testpop"), + "description": description or "Test Role", + "user_ids": json.dumps(user_ids), + } + role_response = self.galaxy_interactor.post("roles", data=payload, admin=True) + assert role_response.status_code == 200 + return role_response.json()[0] + + def make_private(self, history_id, dataset_id): + role_id = self.user_private_role_id() + # Give manage permission to the user. + payload = { + "access": json.dumps([role_id]), + "manage": json.dumps([role_id]), + } + url = "histories/%s/contents/%s/permissions" % (history_id, dataset_id) + update_response = self.galaxy_interactor._put(url, payload, admin=True) + assert update_response.status_code == 200, update_response.content + return update_response.json() + + def get_random_name(self, prefix=None, suffix=None, len=10): + # stolen from navigates_galaxy.py + return '%s%s%s' % ( + prefix or '', + ''.join(random.choice(string.ascii_lowercase + string.digits) for _ in range(len)), + suffix or '', + ) + class DatasetPopulator(BaseDatasetPopulator): @@ -577,20 +633,16 @@ class LibraryPopulator(object): LIBRARY_ADD_in=perm_list, LIBRARY_MANAGE_in=perm_list, ) - self.galaxy_interactor.post("libraries/%s/permissions" % library_id, data=permissions, admin=True) + response = self.galaxy_interactor.post("libraries/%s/permissions" % library_id, data=permissions, admin=True) + api_asserts.assert_status_code_is(response, 200) def user_email(self): - users_response = self.galaxy_interactor.get("users") - users = users_response.json() - assert len(users) == 1 - return users[0]["email"] + # deprecated - use DatasetPopulator + return self.dataset_populator.user_email() def user_private_role_id(self): - user_email = self.user_email() - roles_response = self.galaxy_interactor.get("roles", admin=True) - users_roles = [r for r in roles_response.json() if r["name"] == user_email] - assert len(users_roles) == 1 - return users_roles[0]["id"] + # deprecated - use DatasetPopulator + return self.dataset_populator.user_private_role_id() def create_dataset_request(self, library, **kwds): upload_option = kwds.get("upload_option", "upload_file") diff --git a/test/unit/managers/test_DatasetManager.py b/test/unit/managers/test_DatasetManager.py index d4e0571bc35..aec1bd31a59 100644 --- a/test/unit/managers/test_DatasetManager.py +++ b/test/unit/managers/test_DatasetManager.py @@ -10,10 +10,8 @@ from galaxy import ( exceptions, model ) -from galaxy.managers import rbac_secured from galaxy.managers.base import SkipAttribute from galaxy.managers.datasets import ( - DatasetDeserializer, DatasetManager, DatasetSerializer ) @@ -317,123 +315,6 @@ class DatasetSerializerTestCase(BaseTestCase): self.assertIsJsonifyable(serialized) -# ============================================================================= -class DatasetDeserializerTestCase(BaseTestCase): - - def set_up_managers(self): - super(DatasetDeserializerTestCase, self).set_up_managers() - self.dataset_manager = DatasetManager(self.app) - self.dataset_serializer = DatasetSerializer(self.app) - self.dataset_deserializer = DatasetDeserializer(self.app) - self.role_manager = RoleManager(self.app) - - def test_deserialize_delete(self): - dataset = self.dataset_manager.create() - - self.log('should raise when deserializing deleted from non-bool') - self.assertFalse(dataset.deleted) - self.assertRaises(exceptions.RequestParameterInvalidException, - self.dataset_deserializer.deserialize, dataset, data={'deleted': None}) - self.assertFalse(dataset.deleted) - self.log('should be able to deserialize deleted from True') - self.dataset_deserializer.deserialize(dataset, data={'deleted': True}) - self.assertTrue(dataset.deleted) - self.log('should be able to reverse by deserializing deleted from False') - self.dataset_deserializer.deserialize(dataset, data={'deleted': False}) - self.assertFalse(dataset.deleted) - - def test_deserialize_purge(self): - dataset = self.dataset_manager.create() - - self.log('should raise when deserializing purged from non-bool') - self.assertRaises(exceptions.RequestParameterInvalidException, - self.dataset_deserializer.deserialize, dataset, data={'purged': None}) - self.assertFalse(dataset.purged) - self.log('should be able to deserialize purged from True') - self.dataset_deserializer.deserialize(dataset, data={'purged': True}) - self.assertTrue(dataset.purged) - # TODO: should this raise an error? - self.log('should NOT be able to deserialize purged from False (will remain True)') - self.dataset_deserializer.deserialize(dataset, data={'purged': False}) - self.assertTrue(dataset.purged) - - def test_deserialize_permissions(self): - dataset = self.dataset_manager.create() - who_manages = self.user_manager.create(**user2_data) - self.dataset_manager.permissions.manage.grant(dataset, who_manages) - existing_permissions = self.dataset_serializer.serialize_permissions(dataset, 'permissions', user=who_manages) - existing_manage_permissions = existing_permissions['manage'] - - user3 = self.user_manager.create(**user3_data) - - self.log('deserializing permissions from a non-dictionary should error') - not_a_dict = [] - self.assertRaises(exceptions.RequestParameterInvalidException, self.dataset_deserializer.deserialize, - dataset, user=who_manages, data={'permissions': not_a_dict}) - - self.log('deserializing permissions from a malformed dictionary should error') - self.assertRaises(exceptions.RequestParameterInvalidException, self.dataset_deserializer.deserialize, - dataset, user=who_manages, data={'permissions': dict(nope=[], access=[])}) - - self.log('deserializing permissions with no manage roles should error') - self.assertRaises(exceptions.RequestParameterInvalidException, self.dataset_deserializer.deserialize, - dataset, user=who_manages, data={'permissions': dict(manage=[], access=[])}) - - self.log('deserializing permissions using a non-managing user should error') - self.assertRaises(rbac_secured.DatasetManagePermissionFailedException, self.dataset_deserializer.deserialize, - dataset, user=user3, data={'permissions': existing_permissions}) - - self.log('deserializing permissions using an anon user should error') - self.assertRaises(rbac_secured.DatasetManagePermissionFailedException, self.dataset_deserializer.deserialize, - dataset, user=None, data={'permissions': existing_permissions}) - - self.log('deserializing permissions with a single access should make the dataset private') - private_role = self.user_manager.private_role(who_manages) - private_role = private_role.to_dict(value_mapper={'id': self.app.security.encode_id}) - permissions = dict(manage=existing_manage_permissions, access=[private_role['id']]) - self.dataset_deserializer.deserialize(dataset, user=who_manages, data={ - 'permissions': permissions - }) - self.assertFalse(self.dataset_manager.is_accessible(dataset, user=user3)) - - self.log('deserializing permissions manage should make the permissions available') - self.assertRaises(SkipAttribute, self.dataset_serializer.serialize_permissions, - dataset, 'perms', user=user3) - # now, have who_manages give a manage permission to user3 - private_role = self.user_manager.private_role(user3) - new_manage_permissions = existing_manage_permissions + [self.app.security.encode_id(private_role.id)] - permissions = dict(manage=new_manage_permissions, access=[]) - self.dataset_deserializer.deserialize(dataset, user=who_manages, data={ - 'permissions': permissions - }) - - # deserializing for user3 shouldn't throw a skip bc they can manage - permissions = self.dataset_serializer.serialize_permissions(dataset, 'perms', user=who_manages) - self.assertEqual(new_manage_permissions, permissions['manage']) - - def test_deserialize_permissions_with_admin(self): - dataset = self.dataset_manager.create() - who_manages = self.user_manager.create(**user2_data) - self.dataset_manager.permissions.manage.grant(dataset, who_manages) - existing_permissions = self.dataset_serializer.serialize_permissions(dataset, 'permissions', user=who_manages) - existing_manage_permissions = existing_permissions['manage'] - - user3 = self.user_manager.create(**user3_data) - self.assertRaises(rbac_secured.DatasetManagePermissionFailedException, self.dataset_deserializer.deserialize, - dataset, user=user3, data={'permissions': existing_permissions}) - - self.log('deserializing permissions using an admin user should not error') - private_role = self.user_manager.private_role(who_manages) - private_role = private_role.to_dict(value_mapper={'id' : self.app.security.encode_id}) - permissions = dict(manage=existing_manage_permissions, access=[private_role['id']]) - self.dataset_deserializer.deserialize(dataset, user=who_manages, data={ - 'permissions': permissions - }) - - self.assertRaises(rbac_secured.DatasetManagePermissionFailedException, self.dataset_deserializer.deserialize, - dataset, user=user3, data={'permissions': existing_permissions}) - - # ============================================================================= # NOTE: that we test the DatasetAssociation* classes in either test_HDAManager or test_LDAManager # (as part of those subclasses):