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 sqlalchemy import false, func
from galaxy import util, web
from galaxy.util import inflector
from galaxy.util.hash_util import new_secure_hash
from galaxy.web.form_builder import CheckboxField
from tool_shed.util.web_util import escape
log = logging.getLogger(__name__)
compliance_log = logging.getLogger('COMPLIANCE')
[docs]class Admin(object):
# Override these
user_list_grid = None
role_list_grid = None
group_list_grid = None
delete_operation = None
undelete_operation = None
purge_operation = 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 = "Role '%s' has been renamed to '%s'" % (old_name, 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 = '%s / %s' % (folder.name, folder_path)
if not folder.parent:
root_found = True
else:
folder = folder.parent
folder_path = '%s %s' % (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 += " %s " % 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 = "Role '%s' has not been deleted, so it cannot be undeleted." % role.name
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 += " %s" % 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 = "Role '%s' has not been deleted, so it cannot be purged." % role.name
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 += " %s " % 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 = "Group '%s' has been renamed to '%s'" % (old_name, 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='Role for group %s' % 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 += " %s " % 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 = "Group '%s' has not been deleted, so it cannot be undeleted." % group.name
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 += " %s" % 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 = "Group '%s' has not been deleted, so it cannot be purged." % group.name
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 += " %s " % 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)
if len(password) < 6:
message = "Use a password of at least 6 characters."
status = 'error'
break
elif password != confirm:
message = "Passwords do not match."
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('delete-user-event: %s' % 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 += " %s " % 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 = "User '%s' has not been deleted, so it cannot be undeleted." % user.email
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 += " %s" % 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 = "User '%s' has not been deleted, so it cannot be purged." % user.email
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 += "%s " % 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(q.lower() + "%")):
ac_data = 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 += "Invalid user id (%s) received" % str(user_id)
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("User not found for id (%s)" % 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("Role not found for id (%s)" % 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("Group not found for id (%s)" % str(id))
return group