Source code for galaxy.webapps.galaxy.controllers.admin

import logging

from sqlalchemy import (
    false,
    func,
)

from galaxy import (
    model,
    util,
    web,
)
from galaxy.exceptions import (
    ActionInputError,
    MessageException,
)
from galaxy.managers.quotas import QuotaManager
from galaxy.model import tool_shed_install as install_model
from galaxy.security.validate_user_input import validate_password
from galaxy.tool_shed.util.repository_util import get_ids_of_tool_shed_repositories_being_installed
from galaxy.util import (
    nice_size,
    pretty_print_time_interval,
    sanitize_text,
)
from galaxy.web import url_for
from galaxy.web.framework.helpers import (
    grids,
    time_ago,
)
from galaxy.webapps.base import controller
from tool_shed.util.web_util import escape

log = logging.getLogger(__name__)


[docs]class UserListGrid(grids.Grid):
[docs] class EmailColumn(grids.TextColumn):
[docs] def get_value(self, trans, grid, user): return escape(user.email)
[docs] class UserNameColumn(grids.TextColumn):
[docs] def get_value(self, trans, grid, user): if user.username: return escape(user.username) return "not set"
[docs] class StatusColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, user): if user.purged: return "purged" elif user.deleted: return "deleted" return ""
[docs] class GroupsColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, user): if user.groups: return len(user.groups) return 0
[docs] class RolesColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, user): if user.roles: return len(user.roles) return 0
[docs] class ExternalColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, user): if user.external: return "yes" return "no"
[docs] class LastLoginColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, user): if user.galaxy_sessions: return self.format(user.galaxy_sessions[0].update_time) return "never"
[docs] def sort(self, trans, query, ascending, column_name=None): last_login_subquery = ( trans.sa_session.query( model.GalaxySession.table.c.user_id, func.max(model.GalaxySession.table.c.update_time).label("last_login"), ) .group_by(model.GalaxySession.table.c.user_id) .subquery() ) query = query.outerjoin((last_login_subquery, model.User.table.c.id == last_login_subquery.c.user_id)) if not ascending: query = query.order_by((last_login_subquery.c.last_login).desc().nullslast()) else: query = query.order_by((last_login_subquery.c.last_login).asc().nullsfirst()) return query
[docs] class TimeCreatedColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, user): return user.create_time.strftime("%x")
[docs] class ActivatedColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, user): if user.active: return "Y" else: return "N"
[docs] class DiskUsageColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, user): return user.get_disk_usage(nice_size=True)
[docs] def sort(self, trans, query, ascending, column_name=None): if column_name is None: column_name = self.key column = self.model_class.table.c.get(column_name) if column is None: column = getattr(self.model_class, column_name) if ascending: query = query.order_by(func.coalesce(column, 0).asc()) else: query = query.order_by(func.coalesce(column, 0).desc()) return query
# Grid definition title = "Users" title_id = "users-grid" model_class = model.User default_sort_key = "email" columns = [ EmailColumn( "Email", key="email", link=(lambda item: dict(controller="user", action="information", id=item.id, webapp="galaxy")), attach_popup=True, filterable="advanced", target="top", ), UserNameColumn("User Name", key="username", attach_popup=False, filterable="advanced"), LastLoginColumn("Last Login", format=time_ago, key="last_login", sortable=True), DiskUsageColumn("Disk Usage", key="disk_usage", attach_popup=False), StatusColumn("Status", attach_popup=False, key="deleted"), TimeCreatedColumn("Created", attach_popup=False, key="create_time"), ActivatedColumn("Activated", attach_popup=False, key="active"), GroupsColumn("Groups", attach_popup=False), RolesColumn("Roles", attach_popup=False), ExternalColumn("External", attach_popup=False, key="external"), # Columns that are valid for filtering but are not visible. grids.DeletedColumn("Deleted", key="deleted", visible=False, filterable="advanced"), grids.PurgedColumn("Purged", key="purged", visible=False, filterable="advanced"), ] columns.append( grids.MulticolFilterColumn( "Search", cols_to_filter=[columns[0], columns[1]], key="free-text-search", visible=False, filterable="standard", ) ) global_actions = [grids.GridAction("Create new user", url_args=dict(action="users/create"))] operations = [ grids.GridOperation( "Manage Information", condition=(lambda item: not item.deleted), allow_multiple=False, url_args=dict(controller="user", action="information", webapp="galaxy"), ), grids.GridOperation( "Manage Roles and Groups", condition=(lambda item: not item.deleted), allow_multiple=False, url_args=dict(action="form/manage_roles_and_groups_for_user"), ), grids.GridOperation( "Reset Password", condition=(lambda item: not item.deleted), allow_multiple=True, url_args=dict(action="form/reset_user_password"), target="top", ), grids.GridOperation("Recalculate Disk Usage", condition=(lambda item: not item.deleted), allow_multiple=False), grids.GridOperation("Generate New API Key", allow_multiple=False, async_compatible=True), ] standard_filters = [ grids.GridColumnFilter("Active", args=dict(deleted=False)), grids.GridColumnFilter("Deleted", args=dict(deleted=True, purged=False)), grids.GridColumnFilter("Purged", args=dict(purged=True)), grids.GridColumnFilter("All", args=dict(deleted="All")), ] num_rows_per_page = 50 use_paging = True default_filter = dict(purged="False") use_default_filter = True
[docs] def get_current_item(self, trans, **kwargs): return trans.user
[docs]class RoleListGrid(grids.Grid):
[docs] class NameColumn(grids.TextColumn):
[docs] def get_value(self, trans, grid, role): return escape(role.name)
[docs] class DescriptionColumn(grids.TextColumn):
[docs] def get_value(self, trans, grid, role): if role.description: return escape(role.description) return ""
[docs] class TypeColumn(grids.TextColumn):
[docs] def get_value(self, trans, grid, role): return role.type
[docs] class StatusColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, role): if role.deleted: return "deleted" return ""
[docs] class GroupsColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, role): if role.groups: return len(role.groups) return 0
[docs] class UsersColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, role): if role.users: return len(role.users) return 0
# Grid definition title = "Roles" title_id = "roles-grid" model_class = model.Role default_sort_key = "name" columns = [ NameColumn( "Name", key="name", link=(lambda item: dict(action="form/manage_users_and_groups_for_role", id=item.id, webapp="galaxy")), model_class=model.Role, attach_popup=True, filterable="advanced", target="top", ), DescriptionColumn( "Description", key="description", model_class=model.Role, attach_popup=False, filterable="advanced" ), TypeColumn("Type", key="type", model_class=model.Role, attach_popup=False, filterable="advanced"), GroupsColumn("Groups", attach_popup=False), UsersColumn("Users", attach_popup=False), StatusColumn("Status", attach_popup=False), # Columns that are valid for filtering but are not visible. grids.DeletedColumn("Deleted", key="deleted", visible=False, filterable="advanced"), grids.GridColumn("Last Updated", key="update_time"), ] columns.append( grids.MulticolFilterColumn( "Search", cols_to_filter=[columns[0], columns[1], columns[2]], key="free-text-search", visible=False, filterable="standard", ) ) global_actions = [grids.GridAction("Add new role", url_args=dict(action="form/create_role"))] operations = [ grids.GridOperation( "Edit Name/Description", condition=(lambda item: not item.deleted), allow_multiple=False, url_args=dict(action="form/rename_role"), ), grids.GridOperation( "Edit Permissions", condition=(lambda item: not item.deleted), allow_multiple=False, url_args=dict(action="form/manage_users_and_groups_for_role", webapp="galaxy"), ), grids.GridOperation("Delete", condition=(lambda item: not item.deleted), allow_multiple=True), grids.GridOperation("Undelete", condition=(lambda item: item.deleted), allow_multiple=True), grids.GridOperation("Purge", condition=(lambda item: item.deleted), allow_multiple=True), ] standard_filters = [ grids.GridColumnFilter("Active", args=dict(deleted=False)), grids.GridColumnFilter("Deleted", args=dict(deleted=True)), grids.GridColumnFilter("All", args=dict(deleted="All")), ] num_rows_per_page = 50 use_paging = True
[docs] def apply_query_filter(self, trans, query, **kwargs): return query.filter(model.Role.type != model.Role.types.PRIVATE)
[docs]class GroupListGrid(grids.Grid):
[docs] class NameColumn(grids.TextColumn):
[docs] def get_value(self, trans, grid, group): return escape(group.name)
[docs] class StatusColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, group): if group.deleted: return "deleted" return ""
[docs] class RolesColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, group): if group.roles: return len(group.roles) return 0
[docs] class UsersColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, group): if group.users: return len(group.users) return 0
# Grid definition title = "Groups" title_id = "groups-grid" model_class = model.Group default_sort_key = "name" columns = [ NameColumn( "Name", key="name", link=(lambda item: dict(action="form/manage_users_and_roles_for_group", id=item.id, webapp="galaxy")), model_class=model.Group, attach_popup=True, filterable="advanced", ), UsersColumn("Users", attach_popup=False), RolesColumn("Roles", attach_popup=False), StatusColumn("Status", attach_popup=False), # Columns that are valid for filtering but are not visible. grids.DeletedColumn("Deleted", key="deleted", visible=False, filterable="advanced"), grids.GridColumn("Last Updated", key="update_time", format=pretty_print_time_interval), ] columns.append( grids.MulticolFilterColumn( "Search", cols_to_filter=[columns[0]], key="free-text-search", visible=False, filterable="standard" ) ) global_actions = [grids.GridAction("Add new group", url_args=dict(action="form/create_group"))] operations = [ grids.GridOperation( "Edit Name", condition=(lambda item: not item.deleted), allow_multiple=False, url_args=dict(action="form/rename_group"), ), grids.GridOperation( "Edit Permissions", condition=(lambda item: not item.deleted), allow_multiple=False, url_args=dict(action="form/manage_users_and_roles_for_group", webapp="galaxy"), ), grids.GridOperation("Delete", condition=(lambda item: not item.deleted), allow_multiple=True), grids.GridOperation("Undelete", condition=(lambda item: item.deleted), allow_multiple=True), grids.GridOperation("Purge", condition=(lambda item: item.deleted), allow_multiple=True), ] standard_filters = [ grids.GridColumnFilter("Active", args=dict(deleted=False)), grids.GridColumnFilter("Deleted", args=dict(deleted=True)), grids.GridColumnFilter("All", args=dict(deleted="All")), ] num_rows_per_page = 50 use_paging = True
[docs]class QuotaListGrid(grids.Grid):
[docs] class NameColumn(grids.TextColumn):
[docs] def get_value(self, trans, grid, quota): return escape(quota.name)
[docs] class DescriptionColumn(grids.TextColumn):
[docs] def get_value(self, trans, grid, quota): if quota.description: return escape(quota.description) return ""
[docs] class AmountColumn(grids.TextColumn):
[docs] def get_value(self, trans, grid, quota): return quota.operation + quota.display_amount
[docs] class StatusColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, quota): if quota.deleted: return "deleted" elif quota.default: return f"<strong>default for {quota.default[0].type} users</strong>" return ""
[docs] class UsersColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, quota): if quota.users: return len(quota.users) return 0
[docs] class GroupsColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, quota): if quota.groups: return len(quota.groups) return 0
# Grid definition title = "Quotas" model_class = model.Quota default_sort_key = "name" columns = [ NameColumn( "Name", key="name", link=(lambda item: dict(action="form/edit_quota", id=item.id)), model_class=model.Quota, attach_popup=True, filterable="advanced", ), DescriptionColumn( "Description", key="description", model_class=model.Quota, attach_popup=False, filterable="advanced" ), AmountColumn("Amount", key="amount", model_class=model.Quota, attach_popup=False), UsersColumn("Users", attach_popup=False), GroupsColumn("Groups", attach_popup=False), StatusColumn("Status", attach_popup=False), # Columns that are valid for filtering but are not visible. grids.DeletedColumn("Deleted", key="deleted", visible=False, filterable="advanced"), ] columns.append( grids.MulticolFilterColumn( "Search", cols_to_filter=[columns[0], columns[1]], key="free-text-search", visible=False, filterable="standard", ) ) global_actions = [grids.GridAction("Add new quota", dict(action="form/create_quota"))] operations = [ grids.GridOperation( "Rename", condition=(lambda item: not item.deleted), allow_multiple=False, url_args=dict(action="form/rename_quota"), ), grids.GridOperation( "Change amount", condition=(lambda item: not item.deleted), allow_multiple=False, url_args=dict(action="form/edit_quota"), ), grids.GridOperation( "Manage users and groups", condition=(lambda item: not item.default and not item.deleted), allow_multiple=False, url_args=dict(action="form/manage_users_and_groups_for_quota"), ), grids.GridOperation( "Set as different type of default", condition=(lambda item: item.default), allow_multiple=False, url_args=dict(action="form/set_quota_default"), ), grids.GridOperation( "Set as default", condition=(lambda item: not item.default and not item.deleted), allow_multiple=False, url_args=dict(action="form/set_quota_default"), ), grids.GridOperation( "Unset as default", condition=(lambda item: item.default and not item.deleted), allow_multiple=False ), grids.GridOperation( "Delete", condition=(lambda item: not item.deleted and not item.default), allow_multiple=True ), grids.GridOperation("Undelete", condition=(lambda item: item.deleted), allow_multiple=True), grids.GridOperation("Purge", condition=(lambda item: item.deleted), allow_multiple=True), ] standard_filters = [ grids.GridColumnFilter("Active", args=dict(deleted=False)), grids.GridColumnFilter("Deleted", args=dict(deleted=True)), grids.GridColumnFilter("Purged", args=dict(purged=True)), grids.GridColumnFilter("All", args=dict(deleted="All")), ] num_rows_per_page = 50 use_paging = True
[docs]class ToolVersionListGrid(grids.Grid):
[docs] class ToolIdColumn(grids.TextColumn):
[docs] def get_value(self, trans, grid, tool_version): toolbox = trans.app.toolbox if toolbox.has_tool(tool_version.tool_id, exact=True): link = url_for(controller="tool_runner", tool_id=tool_version.tool_id) link_str = f'<a target="_blank" href="{link}">' return f'<div class="count-box state-color-ok">{link_str}{tool_version.tool_id}</a></div>' return tool_version.tool_id
[docs] class ToolVersionsColumn(grids.TextColumn):
[docs] def get_value(self, trans, grid, tool_version): tool_ids_str = "" toolbox = trans.app.toolbox tool = toolbox._tools_by_id.get(tool_version.tool_id) if tool: for tool_id in tool.lineage.tool_ids: if toolbox.has_tool(tool_id, exact=True): link = url_for(controller="tool_runner", tool_id=tool_id) link_str = f'<a target="_blank" href="{link}">' tool_ids_str += f'<div class="count-box state-color-ok">{link_str}{tool_id}</a></div><br/>' else: tool_ids_str += f"{tool_version.tool_id}<br/>" else: tool_ids_str += f"{tool_version.tool_id}<br/>" return tool_ids_str
# Grid definition title = "Tool versions" model_class = install_model.ToolVersion default_sort_key = "tool_id" columns = [ ToolIdColumn("Tool id", key="tool_id", attach_popup=False), ToolVersionsColumn("Version lineage by tool id (parent/child ordered)"), ] columns.append( grids.MulticolFilterColumn( "Search tool id", cols_to_filter=[columns[0]], key="free-text-search", visible=False, filterable="standard" ) ) num_rows_per_page = 50 use_paging = True
[docs] def build_initial_query(self, trans, **kwd): return trans.install_model.context.query(self.model_class)
[docs]class AdminGalaxy(controller.JSAppLauncher): user_list_grid = UserListGrid() role_list_grid = RoleListGrid() group_list_grid = GroupListGrid() quota_list_grid = QuotaListGrid() tool_version_list_grid = ToolVersionListGrid() delete_operation = grids.GridOperation( "Delete", condition=(lambda item: not item.deleted and not item.purged), allow_multiple=True ) undelete_operation = grids.GridOperation( "Undelete", condition=(lambda item: item.deleted and not item.purged), allow_multiple=True ) purge_operation = grids.GridOperation( "Purge", condition=(lambda item: item.deleted and not item.purged), allow_multiple=True ) impersonate_operation = grids.GridOperation( "Impersonate", url_args=dict(controller="admin", action="impersonate"), condition=(lambda item: not item.deleted and not item.purged), allow_multiple=False, ) activate_operation = grids.GridOperation( "Activate User", condition=(lambda item: not item.active), allow_multiple=False ) resend_activation_email = grids.GridOperation( "Resend Activation Email", condition=(lambda item: not item.active), allow_multiple=False )
[docs] def __init__(self, app): super().__init__(app) self.quota_manager: QuotaManager = QuotaManager(app)
[docs] @web.expose @web.require_admin def index(self, trans, **kwd): return self.client(trans, **kwd)
[docs] @web.expose @web.require_admin def client(self, trans, **kwd): """ Endpoint for admin clientside routes. """ message = escape(kwd.get("message", "")) status = kwd.get("status", "done") settings = { "is_repo_installed": trans.install_model.context.query(trans.install_model.ToolShedRepository).first() is not None, "installing_repository_ids": get_ids_of_tool_shed_repositories_being_installed(trans.app, as_string=True), "is_tool_shed_installed": bool(trans.app.tool_shed_registry and trans.app.tool_shed_registry.tool_sheds), } return self._bootstrapped_client(trans, app_name="admin", settings=settings, message=message, status=status)
[docs] @web.expose @web.json @web.require_admin def data_tables_list(self, trans, **kwd): data = [] message = kwd.get("message", "") status = kwd.get("status", "done") sorted_data_tables = sorted(trans.app.tool_data_tables.get_tables().items()) for _data_table_elem_name, data_table in sorted_data_tables: for filename, file_dict in data_table.filenames.items(): file_missing = ["file missing"] if not file_dict.get("found") else [] data.append( { "name": data_table.name, "filename": filename, "tool_data_path": file_dict.get("tool_data_path"), "errors": ", ".join(file_missing + [error for error in file_dict.get("errors", [])]), } ) return {"data": data, "message": message, "status": status}
[docs] @web.expose @web.json @web.require_admin def data_types_list(self, trans, **kwd): datatypes = [] keys = set() message = kwd.get("message", "") status = kwd.get("status", "done") for dtype in sorted(trans.app.datatypes_registry.datatype_elems, key=lambda dt: dt.get("extension")): attrib = dict(dtype.attrib) datatypes.append(attrib) keys |= set(attrib.keys()) return {"keys": list(keys), "data": datatypes, "message": message, "status": status}
[docs] @web.expose @web.json @web.require_admin def users_list(self, trans, **kwd): message = kwd.get("message", "") status = kwd.get("status", "") if "operation" in kwd: id = kwd.get("id") if not id: message, status = (f"Invalid user id ({str(id)}) received.", "error") ids = util.listify(id) operation = kwd["operation"].lower() if operation == "delete": message, status = self._delete_user(trans, ids) elif operation == "undelete": message, status = self._undelete_user(trans, ids) elif operation == "purge": message, status = self._purge_user(trans, ids) elif operation == "recalculate disk usage": message, status = self._recalculate_user(trans, id) elif operation == "generate new api key": message, status = self._new_user_apikey(trans, id) elif operation == "activate user": message, status = self._activate_user(trans, id) elif operation == "resend activation email": message, status = self._resend_activation_email(trans, id) if message and status: kwd["message"] = util.sanitize_text(message) kwd["status"] = status 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) if trans.app.config.allow_user_impersonation: if self.impersonate_operation not in self.user_list_grid.operations: self.user_list_grid.operations.append(self.impersonate_operation) if trans.app.config.user_activation_on: if self.activate_operation not in self.user_list_grid.operations: self.user_list_grid.operations.append(self.activate_operation) self.user_list_grid.operations.append(self.resend_activation_email) return self.user_list_grid(trans, **kwd)
[docs] @web.legacy_expose_api @web.require_admin def quotas_list(self, trans, payload=None, **kwargs): message = kwargs.get("message", "") status = kwargs.get("status", "") if "operation" in kwargs: id = kwargs.get("id") if not id: return self.message_exception(trans, f"Invalid quota id ({str(id)}) received.") quotas = [] for quota_id in util.listify(id): try: quotas.append(get_quota(trans, quota_id)) except MessageException as e: return self.message_exception(trans, util.unicodify(e)) operation = kwargs.pop("operation").lower() try: if operation == "delete": message = self.quota_manager.delete_quota(quotas) elif operation == "undelete": message = self.quota_manager.undelete_quota(quotas) elif operation == "purge": message = self.quota_manager.purge_quota(quotas) elif operation == "unset as default": message = self.quota_manager.unset_quota_default(quotas[0]) except ActionInputError as e: message, status = (e.err_msg, "error") if message: kwargs["message"] = util.sanitize_text(message) kwargs["status"] = status or "done" return self.quota_list_grid(trans, **kwargs)
[docs] @web.legacy_expose_api @web.require_admin def create_quota(self, trans, payload=None, **kwd): if trans.request.method == "GET": all_users = [] all_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) ): all_users.append((user.email, trans.security.encode_id(user.id))) for group in ( trans.sa_session.query(trans.app.model.Group) .filter(trans.app.model.Group.deleted == false()) .order_by(trans.app.model.Group.name) ): all_groups.append((group.name, trans.security.encode_id(group.id))) default_options = [("No", "no")] for type_ in trans.app.model.DefaultQuotaAssociation.types: default_options.append((f"Yes, {type_}", type_)) return { "title": "Create Quota", "inputs": [ {"name": "name", "label": "Name"}, {"name": "description", "label": "Description"}, {"name": "amount", "label": "Amount", "help": 'Examples: "10000MB", "99 gb", "0.2T", "unlimited"'}, { "name": "operation", "label": "Assign, increase by amount, or decrease by amount?", "options": [("=", "="), ("+", "+"), ("-", "-")], }, { "name": "default", "label": "Is this quota a default for a class of users (if yes, what type)?", "options": default_options, "help": "Warning: Any users or groups associated with this quota will be disassociated.", }, build_select_input("in_groups", "Groups", all_groups, []), build_select_input("in_users", "Users", all_users, []), ], } else: try: quota, message = self.quota_manager.create_quota(payload, decode_id=trans.security.decode_id) return {"message": message} except ActionInputError as e: return self.message_exception(trans, e.err_msg)
[docs] @web.legacy_expose_api @web.require_admin def rename_quota(self, trans, payload=None, **kwd): id = kwd.get("id") if not id: return self.message_exception(trans, "No quota id received for renaming.") quota = get_quota(trans, id) if trans.request.method == "GET": return { "title": "Change quota name and description for '%s'" % util.sanitize_text(quota.name), "inputs": [ {"name": "name", "label": "Name", "value": quota.name}, {"name": "description", "label": "Description", "value": quota.description}, ], } else: try: return {"message": self.quota_manager.rename_quota(quota, util.Params(payload))} except ActionInputError as e: return self.message_exception(trans, e.err_msg)
[docs] @web.legacy_expose_api @web.require_admin def manage_users_and_groups_for_quota(self, trans, payload=None, **kwd): quota_id = kwd.get("id") if not quota_id: return self.message_exception(trans, f"Invalid quota id ({str(quota_id)}) received") quota = get_quota(trans, quota_id) if trans.request.method == "GET": in_users = [] all_users = [] in_groups = [] all_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 quota.users]: in_users.append(trans.security.encode_id(user.id)) all_users.append((user.email, trans.security.encode_id(user.id))) for group in ( trans.sa_session.query(trans.app.model.Group) .filter(trans.app.model.Group.deleted == false()) .order_by(trans.app.model.Group.name) ): if group in [x.group for x in quota.groups]: in_groups.append(trans.security.encode_id(group.id)) all_groups.append((group.name, trans.security.encode_id(group.id))) return { "title": "Quota '%s'" % quota.name, "message": "Quota '%s' is currently associated with %d user(s) and %d group(s)." % (quota.name, len(in_users), len(in_groups)), "status": "info", "inputs": [ build_select_input("in_groups", "Groups", all_groups, in_groups), build_select_input("in_users", "Users", all_users, in_users), ], } else: try: return { "message": self.quota_manager.manage_users_and_groups_for_quota( quota, util.Params(payload), decode_id=trans.security.decode_id ) } except ActionInputError as e: return self.message_exception(trans, e.err_msg)
[docs] @web.legacy_expose_api @web.require_admin def edit_quota(self, trans, payload=None, **kwd): id = kwd.get("id") if not id: return self.message_exception(trans, "No quota id received for renaming.") quota = get_quota(trans, id) if trans.request.method == "GET": return { "title": "Edit quota size for '%s'" % util.sanitize_text(quota.name), "inputs": [ { "name": "amount", "label": "Amount", "value": quota.display_amount, "help": 'Examples: "10000MB", "99 gb", "0.2T", "unlimited"', }, { "name": "operation", "label": "Assign, increase by amount, or decrease by amount?", "options": [("=", "="), ("+", "+"), ("-", "-")], "value": quota.operation, }, ], } else: try: return {"message": self.quota_manager.edit_quota(quota, util.Params(payload))} except ActionInputError as e: return self.message_exception(trans, e.err_msg)
[docs] @web.legacy_expose_api @web.require_admin def set_quota_default(self, trans, payload=None, **kwd): id = kwd.get("id") if not id: return self.message_exception(trans, "No quota id received for renaming.") quota = get_quota(trans, id) if trans.request.method == "GET": default_value = quota.default[0].type if quota.default else "no" default_options = [("No", "no")] for typ in trans.app.model.DefaultQuotaAssociation.types.__members__.values(): default_options.append((f"Yes, {typ}", typ)) return { "title": "Set quota default for '%s'" % util.sanitize_text(quota.name), "inputs": [ { "name": "default", "label": "Is this quota a default for a class of users (if yes, what type)?", "options": default_options, "value": default_value, "help": "Warning: Any users or groups associated with this quota will be disassociated.", } ], } else: try: return {"message": self.quota_manager.set_quota_default(quota, util.Params(payload))} except ActionInputError as e: return self.message_exception(trans, e.err_msg)
[docs] @web.expose @web.require_admin def impersonate(self, trans, **kwd): if not trans.app.config.allow_user_impersonation: return trans.show_error_message("User impersonation is not enabled in this instance of Galaxy.") user = None user_id = kwd.get("id", None) if user_id is not None: try: user = trans.sa_session.query(trans.app.model.User).get(trans.security.decode_id(user_id)) if user: trans.handle_user_logout() trans.handle_user_login(user) return trans.show_message( f"You are now logged in as {user.email}, <a target=\"_top\" href=\"{url_for(controller='root')}\">return to the home page</a>", use_panels=True, ) except Exception: log.exception("Error fetching user for impersonation") return trans.response.send_redirect( web.url_for(controller="admin", action="users", message="Invalid user selected", status="error") )
[docs] @web.legacy_expose_api @web.require_admin def tool_versions_list(self, trans, **kwd): return self.tool_version_list_grid(trans, **kwd)
[docs] @web.expose @web.json @web.require_admin def roles_list(self, trans, **kwargs): message = kwargs.get("message") status = kwargs.get("status") if "operation" in kwargs: id = kwargs.get("id", None) if not id: message, status = (f"Invalid role id ({str(id)}) received.", "error") ids = util.listify(id) operation = kwargs["operation"].lower().replace("+", " ") if operation == "delete": message, status = self._delete_role(trans, ids) elif operation == "undelete": message, status = self._undelete_role(trans, ids) elif operation == "purge": message, status = self._purge_role(trans, ids) if message and status: kwargs["message"] = util.sanitize_text(message) kwargs["status"] = status return self.role_list_grid(trans, **kwargs)
[docs] @web.legacy_expose_api @web.require_admin def create_role(self, trans, payload=None, **kwd): if trans.request.method == "GET": all_users = [] all_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) ): all_users.append((user.email, trans.security.encode_id(user.id))) for group in ( trans.sa_session.query(trans.app.model.Group) .filter(trans.app.model.Group.deleted == false()) .order_by(trans.app.model.Group.name) ): all_groups.append((group.name, trans.security.encode_id(group.id))) return { "title": "Create Role", "inputs": [ {"name": "name", "label": "Name"}, {"name": "description", "label": "Description"}, build_select_input("in_groups", "Groups", all_groups, []), build_select_input("in_users", "Users", all_users, []), { "name": "auto_create", "label": "Create a new group of the same name for this role:", "type": "boolean", "optional": True, }, ], } else: name = util.restore_text(payload.get("name", "")) description = util.restore_text(payload.get("description", "")) auto_create_checked = payload.get("auto_create") in_users = [ trans.sa_session.query(trans.app.model.User).get(trans.security.decode_id(x)) for x in util.listify(payload.get("in_users")) ] in_groups = [ trans.sa_session.query(trans.app.model.Group).get(trans.security.decode_id(x)) for x in util.listify(payload.get("in_groups")) ] if not name or not description: return self.message_exception(trans, "Enter a valid name and a description.") elif trans.sa_session.query(trans.app.model.Role).filter(trans.app.model.Role.name == name).first(): return self.message_exception( trans, "Role names must be unique and a role with that name already exists, so choose another name." ) elif None in in_users or None in in_groups: return self.message_exception(trans, "One or more invalid user/group id has been provided.") 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 in_users: ura = trans.app.model.UserRoleAssociation(user, role) trans.sa_session.add(ura) # Create the GroupRoleAssociations for group in in_groups: gra = trans.app.model.GroupRoleAssociation(group, role) trans.sa_session.add(gra) if auto_create_checked: # Check if role with same name already exists if trans.sa_session.query(trans.app.model.Group).filter(trans.app.model.Group.name == name).first(): return self.message_exception( trans, "A group with that name already exists, so choose another name or disable group creation.", ) # 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 = f"Role '{role.name}' has been created with {len(in_users)} associated users and {num_in_groups} associated groups." if auto_create_checked: message += ( "One of the groups associated with this role is the newly created group with the same name." ) return {"message": message}
[docs] @web.legacy_expose_api @web.require_admin def rename_role(self, trans, payload=None, **kwd): id = kwd.get("id") if not id: return self.message_exception(trans, "No role id received for renaming.") role = get_role(trans, id) if trans.request.method == "GET": return { "title": "Change role name and description for '%s'" % util.sanitize_text(role.name), "inputs": [ {"name": "name", "label": "Name", "value": role.name}, {"name": "description", "label": "Description", "value": role.description}, ], } else: old_name = role.name new_name = util.restore_text(payload.get("name")) new_description = util.restore_text(payload.get("description")) if not new_name: return self.message_exception(trans, "Enter a valid role name.") else: existing_role = ( trans.sa_session.query(trans.app.model.Role).filter(trans.app.model.Role.name == new_name).first() ) if existing_role and existing_role.id != role.id: return self.message_exception(trans, "A role with that name already exists.") 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() return {"message": f"Role '{old_name}' has been renamed to '{new_name}'."}
[docs] @web.legacy_expose_api @web.require_admin def manage_users_and_groups_for_role(self, trans, payload=None, **kwd): role_id = kwd.get("id") if not role_id: return self.message_exception(trans, f"Invalid role id ({str(role_id)}) received") role = get_role(trans, role_id) if trans.request.method == "GET": in_users = [] all_users = [] in_groups = [] all_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(trans.security.encode_id(user.id)) all_users.append((user.email, trans.security.encode_id(user.id))) for group in ( trans.sa_session.query(trans.app.model.Group) .filter(trans.app.model.Group.deleted == false()) .order_by(trans.app.model.Group.name) ): if group in [x.group for x in role.groups]: in_groups.append(trans.security.encode_id(group.id)) all_groups.append((group.name, trans.security.encode_id(group.id))) return { "title": "Role '%s'" % role.name, "message": "Role '%s' is currently associated with %d user(s) and %d group(s)." % (role.name, len(in_users), len(in_groups)), "status": "info", "inputs": [ build_select_input("in_groups", "Groups", all_groups, in_groups), build_select_input("in_users", "Users", all_users, in_users), ], } else: in_users = [ trans.sa_session.query(trans.app.model.User).get(trans.security.decode_id(x)) for x in util.listify(payload.get("in_users")) ] in_groups = [ trans.sa_session.query(trans.app.model.Group).get(trans.security.decode_id(x)) for x in util.listify(payload.get("in_groups")) ] if None in in_users or None in in_groups: return self.message_exception(trans, "One or more invalid user/group id has been provided.") 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() trans.app.security_agent.set_entity_role_associations(roles=[role], users=in_users, groups=in_groups) trans.sa_session.refresh(role) return { "message": f"Role '{role.name}' has been updated with {len(in_users)} associated users and {len(in_groups)} associated groups." }
def _delete_role(self, trans, ids): 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} " return (message, "done") def _undelete_role(self, trans, ids): count = 0 undeleted_roles = "" for role_id in ids: role = get_role(trans, role_id) if not role.deleted: return (f"Role '{role.name}' has not been deleted, so it cannot be undeleted.", "error") role.deleted = False trans.sa_session.add(role) trans.sa_session.flush() count += 1 undeleted_roles += f" {role.name}" return ("Undeleted %d roles: %s" % (count, undeleted_roles), "done") def _purge_role(self, trans, ids): # 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 message = "Purged %d roles: " % len(ids) for role_id in ids: role = get_role(trans, role_id) if not role.deleted: return (f"Role '{role.name}' has not been deleted, so it cannot be purged.", "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} " return (message, "done")
[docs] @web.legacy_expose_api @web.require_admin def groups_list(self, trans, **kwargs): message = kwargs.get("message") status = kwargs.get("status") if "operation" in kwargs: id = kwargs.get("id") if not id: return self.message_exception(trans, f"Invalid group id ({str(id)}) received.") ids = util.listify(id) operation = kwargs["operation"].lower().replace("+", " ") if operation == "delete": message, status = self._delete_group(trans, ids) elif operation == "undelete": message, status = self._undelete_group(trans, ids) elif operation == "purge": message, status = self._purge_group(trans, ids) if message and status: kwargs["message"] = util.sanitize_text(message) kwargs["status"] = status return self.group_list_grid(trans, **kwargs)
[docs] @web.legacy_expose_api @web.require_admin def rename_group(self, trans, payload=None, **kwd): id = kwd.get("id") if not id: return self.message_exception(trans, "No group id received for renaming.") group = get_group(trans, id) if trans.request.method == "GET": return { "title": "Change group name for '%s'" % util.sanitize_text(group.name), "inputs": [{"name": "name", "label": "Name", "value": group.name}], } else: old_name = group.name new_name = util.restore_text(payload.get("name")) if not new_name: return self.message_exception(trans, "Enter a valid group name.") else: existing_group = ( trans.sa_session.query(trans.app.model.Group).filter(trans.app.model.Group.name == new_name).first() ) if existing_group and existing_group.id != group.id: return self.message_exception(trans, "A group with that name already exists.") else: if not (group.name == new_name): group.name = new_name trans.sa_session.add(group) trans.sa_session.flush() return {"message": f"Group '{old_name}' has been renamed to '{new_name}'."}
[docs] @web.legacy_expose_api @web.require_admin def manage_users_and_roles_for_group(self, trans, payload=None, **kwd): group_id = kwd.get("id") if not group_id: return self.message_exception(trans, f"Invalid group id ({str(group_id)}) received") group = get_group(trans, group_id) if trans.request.method == "GET": in_users = [] all_users = [] in_roles = [] all_roles = [] 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(trans.security.encode_id(user.id)) all_users.append((user.email, trans.security.encode_id(user.id))) for role in ( trans.sa_session.query(trans.app.model.Role) .filter(trans.app.model.Role.deleted == false()) .order_by(trans.app.model.Role.name) ): if role in [x.role for x in group.roles]: in_roles.append(trans.security.encode_id(role.id)) all_roles.append((role.name, trans.security.encode_id(role.id))) return { "title": "Group '%s'" % group.name, "message": "Group '%s' is currently associated with %d user(s) and %d role(s)." % (group.name, len(in_users), len(in_roles)), "status": "info", "inputs": [ build_select_input("in_roles", "Roles", all_roles, in_roles), build_select_input("in_users", "Users", all_users, in_users), ], } else: in_users = [ trans.sa_session.query(trans.app.model.User).get(trans.security.decode_id(x)) for x in util.listify(payload.get("in_users")) ] in_roles = [ trans.sa_session.query(trans.app.model.Role).get(trans.security.decode_id(x)) for x in util.listify(payload.get("in_roles")) ] if None in in_users or None in in_roles: return self.message_exception(trans, "One or more invalid user/role id has been provided.") trans.app.security_agent.set_entity_group_associations(groups=[group], users=in_users, roles=in_roles) trans.sa_session.refresh(group) return { "message": f"Group '{group.name}' has been updated with {len(in_users)} associated users and {len(in_roles)} associated roles." }
[docs] @web.legacy_expose_api @web.require_admin def create_group(self, trans, payload=None, **kwd): if trans.request.method == "GET": all_users = [] all_roles = [] 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) ): all_users.append((user.email, trans.security.encode_id(user.id))) for role in ( trans.sa_session.query(trans.app.model.Role) .filter(trans.app.model.Role.deleted == false()) .order_by(trans.app.model.Role.name) ): all_roles.append((role.name, trans.security.encode_id(role.id))) return { "title": "Create Group", "title_id": "create-group", "inputs": [ {"name": "name", "label": "Name"}, build_select_input("in_roles", "Roles", all_roles, []), build_select_input("in_users", "Users", all_users, []), { "name": "auto_create", "label": "Create a new role of the same name for this group:", "type": "boolean", "optional": True, }, ], } else: name = util.restore_text(payload.get("name", "")) auto_create_checked = payload.get("auto_create") in_users = [ trans.sa_session.query(trans.app.model.User).get(trans.security.decode_id(x)) for x in util.listify(payload.get("in_users")) ] in_roles = [ trans.sa_session.query(trans.app.model.Role).get(trans.security.decode_id(x)) for x in util.listify(payload.get("in_roles")) ] if not name: return self.message_exception(trans, "Enter a valid name.") elif trans.sa_session.query(trans.app.model.Group).filter(trans.app.model.Group.name == name).first(): return self.message_exception( trans, "Group names must be unique and a group with that name already exists, so choose another name.", ) elif None in in_users or None in in_roles: return self.message_exception(trans, "One or more invalid user/role id has been provided.") else: # Create the role group = trans.app.model.Group(name=name) trans.sa_session.add(group) # Create the UserRoleAssociations for user in in_users: uga = trans.app.model.UserGroupAssociation(user, group) trans.sa_session.add(uga) # Create the GroupRoleAssociations for role in in_roles: gra = trans.app.model.GroupRoleAssociation(group, role) trans.sa_session.add(gra) if auto_create_checked: # Check if role with same name already exists if trans.sa_session.query(trans.app.model.Role).filter(trans.app.model.Role.name == name).first(): return self.message_exception( trans, "A role with that name already exists, so choose another name or disable role creation.", ) # Create the role role = trans.app.model.Role(name=name, description=f"Role for group {name}") trans.sa_session.add(role) # Associate the group with the role 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 auto_create_checked: message += ( "One of the roles associated with this group is the newly created role with the same name." ) return {"message": message}
def _delete_group(self, trans, ids): 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} " return (message, "done") def _undelete_group(self, trans, ids): count = 0 undeleted_groups = "" for group_id in ids: group = get_group(trans, group_id) if not group.deleted: return (f"Group '{group.name}' has not been deleted, so it cannot be undeleted.", "error") group.deleted = False trans.sa_session.add(group) trans.sa_session.flush() count += 1 undeleted_groups += f" {group.name}" return ("Undeleted %d groups: %s" % (count, undeleted_groups), "done") def _purge_group(self, trans, ids): message = "Purged %d groups: " % len(ids) for group_id in ids: group = get_group(trans, group_id) if not group.deleted: return (f"Group '{group.name}' has not been deleted, so it cannot be purged.", "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} " return (message, "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.legacy_expose_api @web.require_admin def reset_user_password(self, trans, payload=None, **kwd): users = {user_id: get_user(trans, user_id) for user_id in util.listify(kwd.get("id"))} if users: if trans.request.method == "GET": return { "message": f"Changes password(s) for: {', '.join(user.email for user in users.values())}.", "status": "info", "inputs": [ {"name": "password", "label": "New password", "type": "password"}, {"name": "confirm", "label": "Confirm password", "type": "password"}, ], } else: password = payload.get("password") confirm = payload.get("confirm") message = validate_password(trans, password, confirm) if message: return self.message_exception(trans, message) for user in users.values(): user.set_password_cleartext(password) trans.sa_session.add(user) trans.sa_session.flush() return {"message": "Passwords reset for %d user(s)." % len(users)} else: return self.message_exception(trans, "Please specify user ids.")
def _delete_user(self, trans, ids): message = "Deleted %d users: " % len(ids) for user_id in ids: user = get_user(trans, user_id) # Actually do the delete self.user_manager.delete(user) # Accumulate messages for the return message message += f" {user.email} " return (message, "done") def _undelete_user(self, trans, ids): count = 0 undeleted_users = "" for user_id in ids: user = get_user(trans, user_id) # Actually do the undelete self.user_manager.undelete(user) # Count and accumulate messages to return to the admin panel count += 1 undeleted_users += f" {user.email}" message = "Undeleted %d users: %s" % (count, undeleted_users) return (message, "done") def _purge_user(self, trans, ids): # 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 # - 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 message = "Purged %d users: " % len(ids) for user_id in ids: user = get_user(trans, user_id) self.user_manager.purge(user) message += f"\t{user.email}\n " return (message, "done") def _recalculate_user(self, trans, user_id): user = trans.sa_session.query(trans.model.User).get(trans.security.decode_id(user_id)) if not user: return (f"User not found for id ({sanitize_text(str(user_id))})", "error") current = user.get_disk_usage() user.calculate_and_set_disk_usage() new = user.get_disk_usage() if new in (current, None): message = f"Usage is unchanged at {nice_size(current)}." else: message = f"Usage has changed by {nice_size(new - current)} to {nice_size(new)}." return (message, "done") def _new_user_apikey(self, trans, user_id): user = trans.sa_session.query(trans.model.User).get(trans.security.decode_id(user_id)) if not user: return (f"User not found for id ({sanitize_text(str(user_id))})", "error") new_key = trans.app.model.APIKeys( user_id=trans.security.decode_id(user_id), key=trans.app.security.get_new_guid() ) trans.sa_session.add(new_key) trans.sa_session.flush() return (f"New key '{new_key.key}' generated for requested user '{user.email}'.", "done") def _activate_user(self, trans, user_id): user = trans.sa_session.query(trans.model.User).get(trans.security.decode_id(user_id)) if not user: return (f"User not found for id ({sanitize_text(str(user_id))})", "error") self.user_manager.activate(user) return (f"Activated user: {user.email}.", "done") def _resend_activation_email(self, trans, user_id): user = trans.sa_session.query(trans.model.User).get(trans.security.decode_id(user_id)) if not user: return (f"User not found for id ({sanitize_text(str(user_id))})", "error") if self.user_manager.send_activation_email(trans, user.email, user.username): return (f"Activation email has been sent to user: {user.email}.", "done") else: return (f"Unable to send activation email to user: {user.email}.", "error")
[docs] @web.legacy_expose_api @web.require_admin def manage_roles_and_groups_for_user(self, trans, payload=None, **kwd): user_id = kwd.get("id") if not user_id: return self.message_exception(trans, f"Invalid user id ({str(user_id)}) received") user = get_user(trans, user_id) if trans.request.method == "GET": in_roles = [] all_roles = [] in_groups = [] all_groups = [] for role in ( trans.sa_session.query(trans.app.model.Role) .filter(trans.app.model.Role.deleted == false()) .order_by(trans.app.model.Role.name) ): if role in [x.role for x in user.roles]: in_roles.append(trans.security.encode_id(role.id)) if 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. all_roles.append((role.name, trans.security.encode_id(role.id))) for group in ( trans.sa_session.query(trans.app.model.Group) .filter(trans.app.model.Group.deleted == false()) .order_by(trans.app.model.Group.name) ): if group in [x.group for x in user.groups]: in_groups.append(trans.security.encode_id(group.id)) all_groups.append((group.name, trans.security.encode_id(group.id))) return { "title": f"Roles and groups for '{user.email}'", "message": f"User '{user.email}' is currently associated with {len(in_roles) - 1} role(s) and is a member of {len(in_groups)} group(s).", "status": "info", "inputs": [ build_select_input("in_roles", "Roles", all_roles, in_roles), build_select_input("in_groups", "Groups", all_groups, in_groups), ], } else: in_roles = [ trans.sa_session.query(trans.app.model.Role).get(trans.security.decode_id(x)) for x in util.listify(payload.get("in_roles")) ] in_groups = [ trans.sa_session.query(trans.app.model.Group).get(trans.security.decode_id(x)) for x in util.listify(payload.get("in_groups")) ] if None in in_groups or None in in_roles: return self.message_exception(trans, "One or more invalid role/group id has been provided.") # make sure the user is not dis-associating himself from his private role private_role = trans.app.security_agent.get_private_user_role(user) if private_role not in in_roles: in_roles.append(private_role) trans.app.security_agent.set_entity_user_associations(users=[user], roles=in_roles, groups=in_groups) trans.sa_session.refresh(user) return { "message": f"User '{user.email}' has been updated with {len(in_roles) - 1} associated roles and {len(in_groups)} associated groups (private roles are not displayed)." }
[docs] @web.expose @web.require_admin def manage_tool_dependencies( self, trans, install_dependencies=False, uninstall_dependencies=False, remove_unused_dependencies=False, selected_tool_ids=None, selected_environments_to_uninstall=None, viewkey="View tool-centric dependencies", ): if not selected_tool_ids: selected_tool_ids = [] if not selected_environments_to_uninstall: selected_environments_to_uninstall = [] tools_by_id = trans.app.toolbox.tools_by_id.copy() view = next(iter(trans.app.toolbox.tools_by_id.values()))._view if selected_tool_ids: # install the dependencies for the tools in the selected_tool_ids list if not isinstance(selected_tool_ids, list): selected_tool_ids = [selected_tool_ids] requirements = {tools_by_id[tid].tool_requirements for tid in selected_tool_ids} if install_dependencies: [view.install_dependencies(r) for r in requirements] elif uninstall_dependencies: [view.uninstall_dependencies(index=None, requirements=r) for r in requirements] if selected_environments_to_uninstall and remove_unused_dependencies: if not isinstance(selected_environments_to_uninstall, list): selected_environments_to_uninstall = [selected_environments_to_uninstall] view.remove_unused_dependency_paths(selected_environments_to_uninstall) return trans.fill_template( "/webapps/galaxy/admin/manage_dependencies.mako", tools=tools_by_id, requirements_status=view.toolbox_requirements_status, tool_ids_by_requirements=view.tool_ids_by_requirements, unused_environments=view.unused_dependency_paths, viewkey=viewkey, )
# ---- Utility methods -------------------------------------------------------
[docs]def build_select_input(name, label, options, value): return { "type": "select", "multiple": True, "optional": True, "individual": True, "name": name, "label": label, "options": options, "value": value, }
[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
[docs]def get_quota(trans, id): """Get a Quota from the database by id.""" # Load user from database id = trans.security.decode_id(id) quota = trans.sa_session.query(trans.model.Quota).get(id) return quota