Source code for galaxy.webapps.galaxy.api.sanitize_allow

"""
API operations allowing clients to retrieve and modify the HTML sanitization allow list.
"""
import logging
from typing import (
    Any,
    Dict,
)

from galaxy import web
from galaxy.webapps.base.controller import BaseAPIController

log = logging.getLogger(__name__)


[docs]class SanitizeAllowController(BaseAPIController):
[docs]    @web.require_admin
    @web.expose_api
    def index(self, trans, **kwd):
        """
        GET /api/sanitize_allow
        Return an object showing the current state of the toolbox and allow list.
        """
        return self._generate_allowlist(trans)

[docs]    @web.require_admin
    @web.expose_api
    def create(self, trans, tool_id, **kwd):
        """
        PUT /api/sanitize_allow
        Add a new tool_id to the allowlist.
        """
        if tool_id not in trans.app.config.sanitize_allowlist:
            trans.app.config.sanitize_allowlist.append(tool_id)
            self._save_allowlist(trans)
        return self._generate_allowlist(trans)

[docs]    @web.require_admin
    @web.expose_api
    def delete(self, trans, tool_id, **kwd):
        """
        DELETE /api/sanitize_allow
        Remove tool_id from allowlist.
        """
        if tool_id in trans.app.config.sanitize_allowlist:
            trans.app.config.sanitize_allowlist.remove(tool_id)
            self._save_allowlist(trans)
        return self._generate_allowlist(trans)

    def _save_allowlist(self, trans):
        trans.app.config.sanitize_allowlist = sorted(trans.app.config.sanitize_allowlist)
        with open(trans.app.config.sanitize_allowlist_file, "w") as f:
            f.write("\n".join(trans.app.config.sanitize_allowlist))
        trans.app.queue_worker.send_control_task("reload_sanitize_allowlist", noop_self=True)

    def _generate_allowlist(self, trans):
        sanitize_dict: Dict[str, Any] = dict(
            blocked_toolshed=[], allowed_toolshed=[], blocked_local=[], allowed_local=[]
        )
        ids = None
        for tool_id in trans.app.config.sanitize_allowlist:
            installed_name = ""
            installed_ids = {"full": "", "allowed": tool_id, "owner": "", "repository": "", "tool": ""}
            for toolbox_id in trans.app.toolbox.tools_by_id:
                if toolbox_id.startswith(tool_id):
                    tool = trans.app.toolbox.tools_by_id[toolbox_id]
                    installed_name = tool.name
                    full_id = tool.id
                    installed_ids = {
                        "full": full_id,
                        "allowed": tool_id,
                        "owner": "/".join(full_id.split("/")[:3]),
                        "repository": "/".join(full_id.split("/")[:4]),
                        "tool": "/".join(full_id.split("/")[:5]),
                    }
                    break
            tool_dict = dict(
                tool_name=installed_name,
                tool_id=tool_id.split("/"),
                ids=installed_ids,
                allowed=True,
                toolshed="/" in tool_id,
            )
            if "/" in tool_id:
                sanitize_dict["allowed_toolshed"].append(tool_dict)
            else:
                sanitize_dict["allowed_local"].append(tool_dict)
        for tool_id in sorted(trans.app.toolbox.tools_by_id):
            if not tool_id.startswith(tuple(trans.app.config.sanitize_allowlist)):
                tool = trans.app.toolbox.tools_by_id[tool_id]
                ids = {
                    "full": tool_id,
                    "owner": "/".join(tool_id.split("/")[:3]),
                    "repository": "/".join(tool_id.split("/")[:4]),
                    "tool": "/".join(tool_id.split("/")[:5]),
                }
                tool_dict = dict(
                    tool_name=tool.name, tool_id=tool_id.split("/"), ids=ids, allowed=False, toolshed="/" in tool_id
                )
                if "/" in tool_id:
                    sanitize_dict["blocked_toolshed"].append(tool_dict)
                else:
                    sanitize_dict["blocked_local"].append(tool_dict)
        return sanitize_dict