"""
MIT License
Copyright (C) 2023 ROCKY4546
https://github.com/rocky4546
This file is part of Cabernet
Permission is hereby granted, free of charge, to any person obtaining a copy of this software
and associated documentation files (the "Software"), to deal in the Software without restriction,
including without limitation the rights to use, copy, modify, merge, publish, distribute,
sublicense, and/or sell copies of the Software, and to permit persons to whom the Software
is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or
substantial portions of the Software.
"""
import logging
import lib.common.utils as utils
import lib.common.exceptions as exceptions
from lib.common.decorators import getrequest
from lib.common.decorators import postrequest
from lib.web.pages.templates import web_templates
from lib.db.db_plugins import DBPlugins
from lib.plugins.plugin_manager.plugin_manager import PluginManager
@getrequest.route('/api/pluginsform')
def get_plugins_form_html(_webserver, _namespace=None, _sort_col=None, _sort_dir=None, filter_dict=None):
plugins_form = PluginsFormHTML(_webserver.config)
_area = _webserver.query_data.get('area')
_plugin = _webserver.query_data.get('plugin')
_repo = _webserver.query_data.get('repo')
if _area is None and _plugin is None and _repo is None:
_webserver.do_mime_response(
404, 'text/html', web_templates['htmlError']
.format('404 - Badly formed request'))
elif _area:
try:
form = plugins_form.get(_area)
_webserver.do_mime_response(200, 'text/html', form)
except exceptions.CabernetException as ex:
_webserver.do_mime_response(
404, 'text/html', web_templates['htmlError']
.format('404 - Badly formed area request'))
elif _plugin and _repo:
try:
form = plugins_form.get_plugin(_repo, _plugin)
_webserver.do_mime_response(200, 'text/html', form)
except exceptions.CabernetException as ex:
_webserver.do_mime_response(
404, 'text/html', web_templates['htmlError']
.format('404 - Badly formed plugin request'))
else:
# case where plugin and repo are not provided together
_webserver.do_mime_response(
404, 'text/html', web_templates['htmlError']
.format('404 - Badly formed plugin/repo request'))
@postrequest.route('/api/pluginsform')
def post_plugins_html(_webserver):
action = _webserver.query_data.get('action')
pluginid = _webserver.query_data.get('pluginId')
repoid = _webserver.query_data.get('repoId')
if action and pluginid and repoid:
action = action[0]
pluginid = pluginid[0]
repoid = repoid[0]
if action == "deletePlugin":
pm = PluginManager(_webserver.plugins)
results = pm.delete_plugin(repoid, pluginid, _webserver.sched_queue)
_webserver.do_mime_response(200, 'text/html', 'STATUS: Deleting plugin: {}:{}
'.format(repoid, pluginid) + str(results))
if action == "addInstance":
pm = PluginManager(_webserver.plugins)
results = pm.add_instance(repoid, pluginid, _webserver.sched_queue)
_webserver.do_mime_response(200, 'text/html', 'STATUS: Adding Instance plugin: {}:{}
'.format(repoid, pluginid) + str(results))
elif action == "installPlugin":
pm = PluginManager(_webserver.plugins)
results = pm.install_plugin(repoid, pluginid, _webserver.sched_queue)
_webserver.do_mime_response(200, 'text/html', 'STATUS: Installing plugin: {}:{}
'.format(repoid, pluginid) + str(results))
elif action == "upgradePlugin":
pm = PluginManager(_webserver.plugins)
results = pm.upgrade_plugin(repoid, pluginid, _webserver.sched_queue)
_webserver.do_mime_response(200, 'text/html', 'STATUS: Installing plugin: {}:{}
'.format(repoid, pluginid) + str(results))
else:
_webserver.do_mime_response(200, 'text/html', "doing something else"+str(action[0]))
else:
_webserver.do_mime_response(
404, 'text/html', web_templates['htmlError']
.format('404 - Badly formed request'))
class PluginsFormHTML:
def __init__(self, _config):
self.logger = logging.getLogger(__name__)
self.config = _config
self.plugin_db = DBPlugins(self.config)
self.active_tab_name = None
self.num_of_plugins = 0
self.plugin_data = None
self.area = None
def get(self, _area):
self.area = _area
return ''.join([self.header, self.body])
def get_plugin(self, _repo_id, _plugin_id):
plugin_defn = self.plugin_db.get_plugins(
_installed=None,
_repo_id=_repo_id,
_plugin_id=_plugin_id)
if not plugin_defn:
self.logger.warning(
'HTTP request: Unknown plugin: {}'
.format(_plugin_id))
raise exceptions.CabernetException(
'Unknown Plugin: {}'
.format(_plugin_id))
plugin_defn = plugin_defn[0]
return ''.join([self.get_plugin_header(plugin_defn), self.get_menu_section(plugin_defn), self.get_plugin_section(plugin_defn)])
def get_menu_top_section(self, _plugin_defn):
return ''.join([
'
',
str(_plugin_defn['changelog']), '