Warning

This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for tool_shed.util.admin_util

import logging
import time
from typing import Optional

from sqlalchemy import (
    false,
    func,
)

from galaxy import (
    util,
    web,
)
from galaxy.security.validate_user_input import validate_password
from galaxy.util import inflector
from galaxy.util.hash_util import new_secure_hash
from galaxy.web.form_builder import CheckboxField
from galaxy.web.legacy_framework.grids import (
    Grid,
    GridOperation,
)
from tool_shed.util.web_util import escape

log = logging.getLogger(__name__)
compliance_log = logging.getLogger("COMPLIANCE")


[docs]class Admin: # Override these user_list_grid: Optional[Grid] = None role_list_grid: Optional[Grid] = None group_list_grid: Optional[Grid] = None delete_operation: Optional[GridOperation] = None undelete_operation: Optional[GridOperation] = None purge_operation: Optional[GridOperation] = None
[docs] @web.expose @web.require_admin def index(self, trans, **kwd): message = escape(kwd.get("message", "")) status = kwd.get("status", "done") return trans.fill_template("/webapps/tool_shed/admin/index.mako", message=message, status=status)
[docs] @web.expose @web.require_admin def center(self, trans, **kwd): message = escape(kwd.get("message", "")) status = kwd.get("status", "done") return trans.fill_template("/webapps/tool_shed/admin/center.mako", message=message, status=status)
[docs] @web.expose @web.require_admin def roles(self, trans, **kwargs): if "operation" in kwargs: operation = kwargs["operation"].lower().replace("+", " ") if operation == "roles": return self.role(trans, **kwargs) if operation == "create": return self.create_role(trans, **kwargs) if operation == "delete": return self.mark_role_deleted(trans, **kwargs) if operation == "undelete": return self.undelete_role(trans, **kwargs) if operation == "purge": return self.purge_role(trans, **kwargs) if operation == "manage users and groups": return self.manage_users_and_groups_for_role(trans, **kwargs) if operation == "manage role associations": # This is currently used only in the Tool Shed. return self.manage_role_associations(trans, **kwargs) if operation == "rename": return self.rename_role(trans, **kwargs) # Render the list view return self.role_list_grid(trans, **kwargs)
[docs] @web.expose @web.require_admin def create_role(self, trans, **kwd): params = util.Params(kwd) message = util.restore_text(params.get("message", "")) status = params.get("status", "done") name = util.restore_text(params.get("name", "")) description = util.restore_text(params.get("description", "")) in_users = util.listify(params.get("in_users", [])) out_users = util.listify(params.get("out_users", [])) in_groups = util.listify(params.get("in_groups", [])) out_groups = util.listify(params.get("out_groups", [])) create_group_for_role = params.get("create_group_for_role", "") create_group_for_role_checked = CheckboxField.is_checked(create_group_for_role) ok = True if params.get("create_role_button", False): if not name or not description: message = "Enter a valid name and a description." status = "error" ok = False elif trans.sa_session.query(trans.app.model.Role).filter(trans.app.model.Role.table.c.name == name).first(): message = "Role names must be unique and a role with that name already exists, so choose another name." status = "error" ok = False else: # Create the role role = trans.app.model.Role(name=name, description=description, type=trans.app.model.Role.types.ADMIN) trans.sa_session.add(role) # Create the UserRoleAssociations for user in [trans.sa_session.query(trans.app.model.User).get(x) for x in in_users]: ura = trans.app.model.UserRoleAssociation(user, role) trans.sa_session.add(ura) # Create the GroupRoleAssociations for group in [trans.sa_session.query(trans.app.model.Group).get(x) for x in in_groups]: gra = trans.app.model.GroupRoleAssociation(group, role) trans.sa_session.add(gra) if create_group_for_role_checked: # Create the group group = trans.app.model.Group(name=name) trans.sa_session.add(group) # Associate the group with the role gra = trans.model.GroupRoleAssociation(group, role) trans.sa_session.add(gra) num_in_groups = len(in_groups) + 1 else: num_in_groups = len(in_groups) trans.sa_session.flush() message = "Role '%s' has been created with %d associated users and %d associated groups. " % ( role.name, len(in_users), num_in_groups, ) if create_group_for_role_checked: message += ( "One of the groups associated with this role is the newly created group with the same name." ) trans.response.send_redirect( web.url_for(controller="admin", action="roles", message=util.sanitize_text(message), status="done") ) if ok: for user in ( trans.sa_session.query(trans.app.model.User) .filter(trans.app.model.User.table.c.deleted == false()) .order_by(trans.app.model.User.table.c.email) ): out_users.append((user.id, user.email)) for group in ( trans.sa_session.query(trans.app.model.Group) .filter(trans.app.model.Group.table.c.deleted == false()) .order_by(trans.app.model.Group.table.c.name) ): out_groups.append((group.id, group.name)) return trans.fill_template( "/webapps/tool_shed/admin/dataset_security/role/role_create.mako", name=name, description=description, in_users=in_users, out_users=out_users, in_groups=in_groups, out_groups=out_groups, create_group_for_role_checked=create_group_for_role_checked, message=message, status=status, )
[docs] @web.expose @web.require_admin def rename_role(self, trans, **kwd): params = util.Params(kwd) message = util.restore_text(params.get("message", "")) status = params.get("status", "done") id = params.get("id", None) if not id: message = "No role ids received for renaming" trans.response.send_redirect( web.url_for(controller="admin", action="roles", message=message, status="error") ) role = get_role(trans, id) if params.get("rename_role_button", False): old_name = role.name new_name = util.restore_text(params.name) new_description = util.restore_text(params.description) if not new_name: message = "Enter a valid name" status = "error" else: existing_role = ( trans.sa_session.query(trans.app.model.Role) .filter(trans.app.model.Role.table.c.name == new_name) .first() ) if existing_role and existing_role.id != role.id: message = "A role with that name already exists" status = "error" else: if not (role.name == new_name and role.description == new_description): role.name = new_name role.description = new_description trans.sa_session.add(role) trans.sa_session.flush() message = f"Role '{old_name}' has been renamed to '{new_name}'" return trans.response.send_redirect( web.url_for( controller="admin", action="roles", message=util.sanitize_text(message), status="done" ) ) return trans.fill_template( "/webapps/tool_shed/admin/dataset_security/role/role_rename.mako", role=role, message=message, status=status )
[docs] @web.expose @web.require_admin def manage_users_and_groups_for_role(self, trans, **kwd): params = util.Params(kwd) message = util.restore_text(params.get("message", "")) status = params.get("status", "done") id = params.get("id", None) if not id: message = "No role ids received for managing users and groups" trans.response.send_redirect( web.url_for(controller="admin", action="roles", message=message, status="error") ) role = get_role(trans, id) if params.get("role_members_edit_button", False): in_users = [trans.sa_session.query(trans.app.model.User).get(x) for x in util.listify(params.in_users)] if trans.webapp.name == "galaxy": for ura in role.users: user = trans.sa_session.query(trans.app.model.User).get(ura.user_id) if user not in in_users: # Delete DefaultUserPermissions for previously associated users that have been removed from the role for dup in user.default_permissions: if role == dup.role: trans.sa_session.delete(dup) # Delete DefaultHistoryPermissions for previously associated users that have been removed from the role for history in user.histories: for dhp in history.default_permissions: if role == dhp.role: trans.sa_session.delete(dhp) trans.sa_session.flush() in_groups = [trans.sa_session.query(trans.app.model.Group).get(x) for x in util.listify(params.in_groups)] trans.app.security_agent.set_entity_role_associations(roles=[role], users=in_users, groups=in_groups) trans.sa_session.refresh(role) message = "Role '%s' has been updated with %d associated users and %d associated groups" % ( role.name, len(in_users), len(in_groups), ) trans.response.send_redirect( web.url_for(controller="admin", action="roles", message=util.sanitize_text(message), status=status) ) in_users = [] out_users = [] in_groups = [] out_groups = [] for user in ( trans.sa_session.query(trans.app.model.User) .filter(trans.app.model.User.table.c.deleted == false()) .order_by(trans.app.model.User.table.c.email) ): if user in [x.user for x in role.users]: in_users.append((user.id, user.email)) else: out_users.append((user.id, user.email)) for group in ( trans.sa_session.query(trans.app.model.Group) .filter(trans.app.model.Group.table.c.deleted == false()) .order_by(trans.app.model.Group.table.c.name) ): if group in [x.group for x in role.groups]: in_groups.append((group.id, group.name)) else: out_groups.append((group.id, group.name)) library_dataset_actions = {} if trans.webapp.name == "galaxy" and len(role.dataset_actions) < 25: # Build a list of tuples that are LibraryDatasetDatasetAssociationss followed by a list of actions # whose DatasetPermissions is associated with the Role # [ ( LibraryDatasetDatasetAssociation [ action, action ] ) ] for dp in role.dataset_actions: for ldda in trans.sa_session.query(trans.app.model.LibraryDatasetDatasetAssociation).filter( trans.app.model.LibraryDatasetDatasetAssociation.dataset_id == dp.dataset_id ): root_found = False folder_path = "" folder = ldda.library_dataset.folder while not root_found: folder_path = f"{folder.name} / {folder_path}" if not folder.parent: root_found = True else: folder = folder.parent folder_path = f"{folder_path} {ldda.name}" library = ( trans.sa_session.query(trans.app.model.Library) .filter(trans.app.model.Library.table.c.root_folder_id == folder.id) .first() ) if library not in library_dataset_actions: library_dataset_actions[library] = {} try: library_dataset_actions[library][folder_path].append(dp.action) except Exception: library_dataset_actions[library][folder_path] = [dp.action] else: message = "Not showing associated datasets, there are too many." status = "info" return trans.fill_template( "/webapps/tool_shed/admin/dataset_security/role/role.mako", role=role, in_users=in_users, out_users=out_users, in_groups=in_groups, out_groups=out_groups, library_dataset_actions=library_dataset_actions, message=message, status=status, )
[docs] @web.expose @web.require_admin def mark_role_deleted(self, trans, **kwd): id = kwd.get("id", None) if not id: message = "No role ids received for deleting" trans.response.send_redirect( web.url_for(controller="admin", action="roles", message=message, status="error") ) ids = util.listify(id) message = "Deleted %d roles: " % len(ids) for role_id in ids: role = get_role(trans, role_id) role.deleted = True trans.sa_session.add(role) trans.sa_session.flush() message += f" {role.name} " trans.response.send_redirect( web.url_for(controller="admin", action="roles", message=util.sanitize_text(message), status="done") )
[docs] @web.expose @web.require_admin def undelete_role(self, trans, **kwd): id = kwd.get("id", None) if not id: message = "No role ids received for undeleting" trans.response.send_redirect( web.url_for(controller="admin", action="roles", message=message, status="error") ) ids = util.listify(id) count = 0 undeleted_roles = "" for role_id in ids: role = get_role(trans, role_id) if not role.deleted: message = f"Role '{role.name}' has not been deleted, so it cannot be undeleted." trans.response.send_redirect( web.url_for(controller="admin", action="roles", message=util.sanitize_text(message), status="error") ) role.deleted = False trans.sa_session.add(role) trans.sa_session.flush() count += 1 undeleted_roles += f" {role.name}" message = "Undeleted %d roles: %s" % (count, undeleted_roles) trans.response.send_redirect( web.url_for(controller="admin", action="roles", message=util.sanitize_text(message), status="done") )
[docs] @web.expose @web.require_admin def purge_role(self, trans, **kwd): # This method should only be called for a Role that has previously been deleted. # Purging a deleted Role deletes all of the following from the database: # - UserRoleAssociations where role_id == Role.id # - DefaultUserPermissions where role_id == Role.id # - DefaultHistoryPermissions where role_id == Role.id # - GroupRoleAssociations where role_id == Role.id # - DatasetPermissionss where role_id == Role.id id = kwd.get("id", None) if not id: message = "No role ids received for purging" trans.response.send_redirect( web.url_for(controller="admin", action="roles", message=util.sanitize_text(message), status="error") ) ids = util.listify(id) message = "Purged %d roles: " % len(ids) for role_id in ids: role = get_role(trans, role_id) if not role.deleted: message = f"Role '{role.name}' has not been deleted, so it cannot be purged." trans.response.send_redirect( web.url_for(controller="admin", action="roles", message=util.sanitize_text(message), status="error") ) # Delete UserRoleAssociations for ura in role.users: user = trans.sa_session.query(trans.app.model.User).get(ura.user_id) # Delete DefaultUserPermissions for associated users for dup in user.default_permissions: if role == dup.role: trans.sa_session.delete(dup) # Delete DefaultHistoryPermissions for associated users for history in user.histories: for dhp in history.default_permissions: if role == dhp.role: trans.sa_session.delete(dhp) trans.sa_session.delete(ura) # Delete GroupRoleAssociations for gra in role.groups: trans.sa_session.delete(gra) # Delete DatasetPermissionss for dp in role.dataset_actions: trans.sa_session.delete(dp) trans.sa_session.flush() message += f" {role.name} " trans.response.send_redirect( web.url_for(controller="admin", action="roles", message=util.sanitize_text(message), status="done") )
[docs] @web.expose @web.require_admin def groups(self, trans, **kwargs): if "operation" in kwargs: operation = kwargs["operation"].lower().replace("+", " ") if operation == "groups": return self.group(trans, **kwargs) if operation == "create": return self.create_group(trans, **kwargs) if operation == "delete": return self.mark_group_deleted(trans, **kwargs) if operation == "undelete": return self.undelete_group(trans, **kwargs) if operation == "purge": return self.purge_group(trans, **kwargs) if operation == "manage users and roles": return self.manage_users_and_roles_for_group(trans, **kwargs) if operation == "rename": return self.rename_group(trans, **kwargs) # Render the list view return self.group_list_grid(trans, **kwargs)
[docs] @web.expose @web.require_admin def rename_group(self, trans, **kwd): params = util.Params(kwd) message = util.restore_text(params.get("message", "")) status = params.get("status", "done") id = params.get("id", None) if not id: message = "No group ids received for renaming" trans.response.send_redirect( web.url_for(controller="admin", action="groups", message=message, status="error") ) group = get_group(trans, id) if params.get("rename_group_button", False): old_name = group.name new_name = util.restore_text(params.name) if not new_name: message = "Enter a valid name" status = "error" else: existing_group = ( trans.sa_session.query(trans.app.model.Group) .filter(trans.app.model.Group.table.c.name == new_name) .first() ) if existing_group and existing_group.id != group.id: message = "A group with that name already exists" status = "error" else: if group.name != new_name: group.name = new_name trans.sa_session.add(group) trans.sa_session.flush() message = f"Group '{old_name}' has been renamed to '{new_name}'" return trans.response.send_redirect( web.url_for( controller="admin", action="groups", message=util.sanitize_text(message), status="done" ) ) return trans.fill_template( "/webapps/tool_shed/admin/dataset_security/group/group_rename.mako", group=group, message=message, status=status, )
[docs] @web.expose @web.require_admin def manage_users_and_roles_for_group(self, trans, **kwd): params = util.Params(kwd) message = util.restore_text(params.get("message", "")) status = params.get("status", "done") group = get_group(trans, params.id) if params.get("group_roles_users_edit_button", False): in_roles = [trans.sa_session.query(trans.app.model.Role).get(x) for x in util.listify(params.in_roles)] in_users = [trans.sa_session.query(trans.app.model.User).get(x) for x in util.listify(params.in_users)] trans.app.security_agent.set_entity_group_associations(groups=[group], roles=in_roles, users=in_users) trans.sa_session.refresh(group) message += "Group '%s' has been updated with %d associated roles and %d associated users" % ( group.name, len(in_roles), len(in_users), ) trans.response.send_redirect( web.url_for(controller="admin", action="groups", message=util.sanitize_text(message), status=status) ) in_roles = [] out_roles = [] in_users = [] out_users = [] for role in ( trans.sa_session.query(trans.app.model.Role) .filter(trans.app.model.Role.table.c.deleted == false()) .order_by(trans.app.model.Role.table.c.name) ): if role in [x.role for x in group.roles]: in_roles.append((role.id, role.name)) else: out_roles.append((role.id, role.name)) for user in ( trans.sa_session.query(trans.app.model.User) .filter(trans.app.model.User.table.c.deleted == false()) .order_by(trans.app.model.User.table.c.email) ): if user in [x.user for x in group.users]: in_users.append((user.id, user.email)) else: out_users.append((user.id, user.email)) message += "Group %s is currently associated with %d roles and %d users" % ( group.name, len(in_roles), len(in_users), ) return trans.fill_template( "/webapps/tool_shed/admin/dataset_security/group/group.mako", group=group, in_roles=in_roles, out_roles=out_roles, in_users=in_users, out_users=out_users, message=message, status=status, )
[docs] @web.expose @web.require_admin def create_group(self, trans, **kwd): params = util.Params(kwd) message = util.restore_text(params.get("message", "")) status = params.get("status", "done") name = util.restore_text(params.get("name", "")) in_users = util.listify(params.get("in_users", [])) out_users = util.listify(params.get("out_users", [])) in_roles = util.listify(params.get("in_roles", [])) out_roles = util.listify(params.get("out_roles", [])) create_role_for_group = params.get("create_role_for_group", "") create_role_for_group_checked = CheckboxField.is_checked(create_role_for_group) ok = True if params.get("create_group_button", False): if not name: message = "Enter a valid name." status = "error" ok = False elif ( trans.sa_session.query(trans.app.model.Group).filter(trans.app.model.Group.table.c.name == name).first() ): message = ( "Group names must be unique and a group with that name already exists, so choose another name." ) status = "error" ok = False else: # Create the group group = trans.app.model.Group(name=name) trans.sa_session.add(group) trans.sa_session.flush() # Create the UserRoleAssociations for user in [trans.sa_session.query(trans.app.model.User).get(x) for x in in_users]: uga = trans.app.model.UserGroupAssociation(user, group) trans.sa_session.add(uga) # Create the GroupRoleAssociations for role in [trans.sa_session.query(trans.app.model.Role).get(x) for x in in_roles]: gra = trans.app.model.GroupRoleAssociation(group, role) trans.sa_session.add(gra) if create_role_for_group_checked: # Create the role role = trans.app.model.Role(name=name, description=f"Role for group {name}") trans.sa_session.add(role) # Associate the role with the group gra = trans.model.GroupRoleAssociation(group, role) trans.sa_session.add(gra) num_in_roles = len(in_roles) + 1 else: num_in_roles = len(in_roles) trans.sa_session.flush() message = "Group '%s' has been created with %d associated users and %d associated roles. " % ( group.name, len(in_users), num_in_roles, ) if create_role_for_group_checked: message += ( "One of the roles associated with this group is the newly created role with the same name." ) trans.response.send_redirect( web.url_for(controller="admin", action="groups", message=util.sanitize_text(message), status="done") ) if ok: for user in ( trans.sa_session.query(trans.app.model.User) .filter(trans.app.model.User.table.c.deleted == false()) .order_by(trans.app.model.User.table.c.email) ): out_users.append((user.id, user.email)) for role in ( trans.sa_session.query(trans.app.model.Role) .filter(trans.app.model.Role.table.c.deleted == false()) .order_by(trans.app.model.Role.table.c.name) ): out_roles.append((role.id, role.name)) return trans.fill_template( "/webapps/tool_shed/admin/dataset_security/group/group_create.mako", name=name, in_users=in_users, out_users=out_users, in_roles=in_roles, out_roles=out_roles, create_role_for_group_checked=create_role_for_group_checked, message=message, status=status, )
[docs] @web.expose @web.require_admin def mark_group_deleted(self, trans, **kwd): params = util.Params(kwd) id = params.get("id", None) if not id: message = "No group ids received for marking deleted" trans.response.send_redirect( web.url_for(controller="admin", action="groups", message=message, status="error") ) ids = util.listify(id) message = "Deleted %d groups: " % len(ids) for group_id in ids: group = get_group(trans, group_id) group.deleted = True trans.sa_session.add(group) trans.sa_session.flush() message += f" {group.name} " trans.response.send_redirect( web.url_for(controller="admin", action="groups", message=util.sanitize_text(message), status="done") )
[docs] @web.expose @web.require_admin def undelete_group(self, trans, **kwd): id = kwd.get("id", None) if not id: message = "No group ids received for undeleting" trans.response.send_redirect( web.url_for(controller="admin", action="groups", message=message, status="error") ) ids = util.listify(id) count = 0 undeleted_groups = "" for group_id in ids: group = get_group(trans, group_id) if not group.deleted: message = f"Group '{group.name}' has not been deleted, so it cannot be undeleted." trans.response.send_redirect( web.url_for( controller="admin", action="groups", message=util.sanitize_text(message), status="error" ) ) group.deleted = False trans.sa_session.add(group) trans.sa_session.flush() count += 1 undeleted_groups += f" {group.name}" message = "Undeleted %d groups: %s" % (count, undeleted_groups) trans.response.send_redirect( web.url_for(controller="admin", action="groups", message=util.sanitize_text(message), status="done") )
[docs] @web.expose @web.require_admin def purge_group(self, trans, **kwd): # This method should only be called for a Group that has previously been deleted. # Purging a deleted Group simply deletes all UserGroupAssociations and GroupRoleAssociations. id = kwd.get("id", None) if not id: message = "No group ids received for purging" trans.response.send_redirect( web.url_for(controller="admin", action="groups", message=util.sanitize_text(message), status="error") ) ids = util.listify(id) message = "Purged %d groups: " % len(ids) for group_id in ids: group = get_group(trans, group_id) if not group.deleted: # We should never reach here, but just in case there is a bug somewhere... message = f"Group '{group.name}' has not been deleted, so it cannot be purged." trans.response.send_redirect( web.url_for( controller="admin", action="groups", message=util.sanitize_text(message), status="error" ) ) # Delete UserGroupAssociations for uga in group.users: trans.sa_session.delete(uga) # Delete GroupRoleAssociations for gra in group.roles: trans.sa_session.delete(gra) trans.sa_session.flush() message += f" {group.name} " trans.response.send_redirect( web.url_for(controller="admin", action="groups", message=util.sanitize_text(message), status="done") )
[docs] @web.expose @web.require_admin def create_new_user(self, trans, **kwd): return trans.response.send_redirect(web.url_for(controller="user", action="create", cntrller="admin"))
[docs] @web.expose @web.require_admin def reset_user_password(self, trans, **kwd): user_id = kwd.get("id", None) if not user_id: message = "No users received for resetting passwords." trans.response.send_redirect( web.url_for(controller="admin", action="users", message=message, status="error") ) user_ids = util.listify(user_id) if "reset_user_password_button" in kwd: message = "" status = "" for user_id in user_ids: user = get_user(trans, user_id) password = kwd.get("password", None) confirm = kwd.get("confirm", None) message = validate_password(trans, password, confirm) if message: status = "error" break else: user.set_password_cleartext(password) trans.sa_session.add(user) trans.sa_session.flush() if not message and not status: message = "Passwords reset for %d %s." % (len(user_ids), inflector.cond_plural(len(user_ids), "user")) status = "done" trans.response.send_redirect( web.url_for(controller="admin", action="users", message=util.sanitize_text(message), status=status) ) users = [get_user(trans, user_id) for user_id in user_ids] if len(user_ids) > 1: user_id = ",".join(user_ids) return trans.fill_template( "/webapps/tool_shed/admin/user/reset_password.mako", id=user_id, users=users, password="", confirm="" )
[docs] @web.expose @web.require_admin def mark_user_deleted(self, trans, **kwd): id = kwd.get("id", None) if not id: message = "No user ids received for deleting" trans.response.send_redirect( web.url_for(controller="admin", action="users", message=message, status="error") ) ids = util.listify(id) message = "Deleted %d users: " % len(ids) for user_id in ids: user = get_user(trans, user_id) user.deleted = True compliance_log.info(f"delete-user-event: {user_id}") # See lib/galaxy/webapps/tool_shed/controllers/admin.py pseudorandom_value = str(int(time.time())) email_hash = new_secure_hash(user.email + pseudorandom_value) uname_hash = new_secure_hash(user.username + pseudorandom_value) for role in user.all_roles(): print( role, self.app.config.redact_username_during_deletion, self.app.config.redact_email_during_deletion ) if self.app.config.redact_username_during_deletion: role.name = role.name.replace(user.username, uname_hash) role.description = role.description.replace(user.username, uname_hash) if self.app.config.redact_email_during_deletion: role.name = role.name.replace(user.email, email_hash) role.description = role.description.replace(user.email, email_hash) if self.app.config.redact_email_during_deletion: user.email = email_hash if self.app.config.redact_username_during_deletion: user.username = uname_hash trans.sa_session.add(user) trans.sa_session.flush() message += f" {user.email} " trans.response.send_redirect( web.url_for(controller="admin", action="users", message=util.sanitize_text(message), status="done") )
[docs] @web.expose @web.require_admin def undelete_user(self, trans, **kwd): id = kwd.get("id", None) if not id: message = "No user ids received for undeleting" trans.response.send_redirect( web.url_for(controller="admin", action="users", message=message, status="error") ) ids = util.listify(id) count = 0 undeleted_users = "" for user_id in ids: user = get_user(trans, user_id) if not user.deleted: message = f"User '{user.email}' has not been deleted, so it cannot be undeleted." trans.response.send_redirect( web.url_for(controller="admin", action="users", message=util.sanitize_text(message), status="error") ) user.deleted = False trans.sa_session.add(user) trans.sa_session.flush() count += 1 undeleted_users += f" {user.email}" message = "Undeleted %d users: %s" % (count, undeleted_users) trans.response.send_redirect( web.url_for(controller="admin", action="users", message=util.sanitize_text(message), status="done") )
[docs] @web.expose @web.require_admin def purge_user(self, trans, **kwd): # This method should only be called for a User that has previously been deleted. # We keep the User in the database ( marked as purged ), and stuff associated # with the user's private role in case we want the ability to unpurge the user # some time in the future. # Purging a deleted User deletes all of the following: # - History where user_id = User.id # - HistoryDatasetAssociation where history_id = History.id # - Dataset where HistoryDatasetAssociation.dataset_id = Dataset.id # - UserGroupAssociation where user_id == User.id # - UserRoleAssociation where user_id == User.id EXCEPT FOR THE PRIVATE ROLE # - UserAddress where user_id == User.id # Purging Histories and Datasets must be handled via the cleanup_datasets.py script id = kwd.get("id", None) if not id: message = "No user ids received for purging" trans.response.send_redirect( web.url_for(controller="admin", action="users", message=util.sanitize_text(message), status="error") ) ids = util.listify(id) message = "Purged %d users: " % len(ids) for user_id in ids: user = get_user(trans, user_id) if not user.deleted: # We should never reach here, but just in case there is a bug somewhere... message = f"User '{user.email}' has not been deleted, so it cannot be purged." trans.response.send_redirect( web.url_for(controller="admin", action="users", message=util.sanitize_text(message), status="error") ) private_role = trans.app.security_agent.get_private_user_role(user) # Delete History for h in user.active_histories: trans.sa_session.refresh(h) for hda in h.active_datasets: # Delete HistoryDatasetAssociation d = trans.sa_session.query(trans.app.model.Dataset).get(hda.dataset_id) # Delete Dataset if not d.deleted: d.deleted = True trans.sa_session.add(d) hda.deleted = True trans.sa_session.add(hda) h.deleted = True trans.sa_session.add(h) # Delete UserGroupAssociations for uga in user.groups: trans.sa_session.delete(uga) # Delete UserRoleAssociations EXCEPT FOR THE PRIVATE ROLE for ura in user.roles: if ura.role_id != private_role.id: trans.sa_session.delete(ura) # Delete UserAddresses for address in user.addresses: trans.sa_session.delete(address) # Purge the user user.purged = True trans.sa_session.add(user) trans.sa_session.flush() message += f"{user.email} " trans.response.send_redirect( web.url_for(controller="admin", action="users", message=util.sanitize_text(message), status="done") )
[docs] @web.expose @web.require_admin def users(self, trans, **kwd): if "operation" in kwd: operation = kwd["operation"].lower() if operation == "roles": return self.user(trans, **kwd) elif operation == "reset password": return self.reset_user_password(trans, **kwd) elif operation == "delete": return self.mark_user_deleted(trans, **kwd) elif operation == "undelete": return self.undelete_user(trans, **kwd) elif operation == "purge": return self.purge_user(trans, **kwd) elif operation == "create": return self.create_new_user(trans, **kwd) elif operation == "manage roles and groups": return self.manage_roles_and_groups_for_user(trans, **kwd) if trans.app.config.allow_user_deletion: if self.delete_operation not in self.user_list_grid.operations: self.user_list_grid.operations.append(self.delete_operation) if self.undelete_operation not in self.user_list_grid.operations: self.user_list_grid.operations.append(self.undelete_operation) if self.purge_operation not in self.user_list_grid.operations: self.user_list_grid.operations.append(self.purge_operation) # Render the list view return self.user_list_grid(trans, **kwd)
[docs] @web.expose @web.require_admin def name_autocomplete_data(self, trans, q=None, limit=None, timestamp=None): """Return autocomplete data for user emails""" ac_data = "" for user in ( trans.sa_session.query(trans.app.model.User) .filter_by(deleted=False) .filter(func.lower(trans.app.model.User.email).like(f"{q.lower()}%")) ): ac_data = f"{ac_data + user.email}\n" return ac_data
[docs] @web.expose @web.require_admin def manage_roles_and_groups_for_user(self, trans, **kwd): user_id = kwd.get("id", None) message = "" status = "" if not user_id: message += f"Invalid user id ({str(user_id)}) received" trans.response.send_redirect( web.url_for(controller="admin", action="users", message=util.sanitize_text(message), status="error") ) user = get_user(trans, user_id) private_role = trans.app.security_agent.get_private_user_role(user) if kwd.get("user_roles_groups_edit_button", False): # Make sure the user is not dis-associating himself from his private role out_roles = kwd.get("out_roles", []) if out_roles: out_roles = [trans.sa_session.query(trans.app.model.Role).get(x) for x in util.listify(out_roles)] if private_role in out_roles: message += "You cannot eliminate a user's private role association. " status = "error" in_roles = kwd.get("in_roles", []) if in_roles: in_roles = [trans.sa_session.query(trans.app.model.Role).get(x) for x in util.listify(in_roles)] out_groups = kwd.get("out_groups", []) if out_groups: out_groups = [trans.sa_session.query(trans.app.model.Group).get(x) for x in util.listify(out_groups)] in_groups = kwd.get("in_groups", []) if in_groups: in_groups = [trans.sa_session.query(trans.app.model.Group).get(x) for x in util.listify(in_groups)] if in_roles: trans.app.security_agent.set_entity_user_associations(users=[user], roles=in_roles, groups=in_groups) trans.sa_session.refresh(user) message += ( "User '%s' has been updated with %d associated roles and %d associated groups (private roles are not displayed)" % (user.email, len(in_roles), len(in_groups)) ) trans.response.send_redirect( web.url_for(controller="admin", action="users", message=util.sanitize_text(message), status="done") ) in_roles = [] out_roles = [] in_groups = [] out_groups = [] for role in ( trans.sa_session.query(trans.app.model.Role) .filter(trans.app.model.Role.table.c.deleted == false()) .order_by(trans.app.model.Role.table.c.name) ): if role in [x.role for x in user.roles]: in_roles.append((role.id, role.name)) elif role.type != trans.app.model.Role.types.PRIVATE: # There is a 1 to 1 mapping between a user and a PRIVATE role, so private roles should # not be listed in the roles form fields, except for the currently selected user's private # role, which should always be in in_roles. The check above is added as an additional # precaution, since for a period of time we were including private roles in the form fields. out_roles.append((role.id, role.name)) for group in ( trans.sa_session.query(trans.app.model.Group) .filter(trans.app.model.Group.table.c.deleted == false()) .order_by(trans.app.model.Group.table.c.name) ): if group in [x.group for x in user.groups]: in_groups.append((group.id, group.name)) else: out_groups.append((group.id, group.name)) message += "User '%s' is currently associated with %d roles and is a member of %d groups" % ( user.email, len(in_roles), len(in_groups), ) if not status: status = "done" return trans.fill_template( "/webapps/tool_shed/admin/user/user.mako", user=user, in_roles=in_roles, out_roles=out_roles, in_groups=in_groups, out_groups=out_groups, message=message, status=status, )
# ---- Utility methods -------------------------------------------------------
[docs]def get_user(trans, user_id): """Get a User from the database by id.""" user = trans.sa_session.query(trans.model.User).get(trans.security.decode_id(user_id)) if not user: return trans.show_error_message(f"User not found for id ({str(user_id)})") return user
[docs]def get_role(trans, id): """Get a Role from the database by id.""" # Load user from database id = trans.security.decode_id(id) role = trans.sa_session.query(trans.model.Role).get(id) if not role: return trans.show_error_message(f"Role not found for id ({str(id)})") return role
[docs]def get_group(trans, id): """Get a Group from the database by id.""" # Load user from database id = trans.security.decode_id(id) group = trans.sa_session.query(trans.model.Group).get(id) if not group: return trans.show_error_message(f"Group not found for id ({str(id)})") return group