Files
FunguyBot/plugins/loadplugin.py

157 lines
6.3 KiB
Python

"""
Plugin for providing a command for the admin to load a plugin.
"""
import os
import logging
import importlib
import simplematrixbotlib as botlib
import sys # Import sys module for unloading plugins
# Dictionary to store loaded plugins
PLUGINS = {}
# Whitelist of allowed plugins to prevent arbitrary code execution
ALLOWED_PLUGINS = {'ai', 'config', 'cron', 'date', 'fortune', 'help', 'isup', 'karma',
'loadplugin', 'plugins', 'proxy', 'sd_text', 'stable-diffusion',
'xkcd', 'youtube-preview', 'youtube-search'}
async def load_plugin(plugin_name):
"""
Asynchronously loads a plugin.
Args:
plugin_name (str): The name of the plugin to load.
Returns:
bool: True if the plugin is loaded successfully, False otherwise.
"""
# Validate plugin name against whitelist
if plugin_name not in ALLOWED_PLUGINS:
logging.error(f"Plugin '{plugin_name}' is not whitelisted")
return False
# Verify that the plugin file exists in the plugins directory
plugin_path = os.path.join("plugins", f"{plugin_name}.py")
if not os.path.isfile(plugin_path):
logging.error(f"Plugin file not found: {plugin_path}")
return False
try:
# Create a mapping of whitelisted plugins to their module paths
plugin_modules = {
'ai': 'plugins.ai',
'config': 'plugins.config',
'cron': 'plugins.cron',
'date': 'plugins.date',
'fortune': 'plugins.fortune',
'help': 'plugins.help',
'isup': 'plugins.isup',
'karma': 'plugins.karma',
'loadplugin': 'plugins.loadplugin',
'plugins': 'plugins.plugins',
'proxy': 'plugins.proxy',
'sd_text': 'plugins.sd_text',
'stable-diffusion': 'plugins.stable-diffusion',
'xkcd': 'plugins.xkcd',
'youtube-preview': 'plugins.youtube-preview',
'youtube-search': 'plugins.youtube-search',
}
# Get the module path from the mapping
module_path = plugin_modules.get(plugin_name)
if not module_path:
logging.error(f"Plugin '{plugin_name}' not found in plugin mapping")
return False
# Import the plugin module using the validated module path
module = importlib.import_module(module_path)
# Add the plugin module to the PLUGINS dictionary
PLUGINS[plugin_name] = module
logging.info(f"Loaded plugin: {plugin_name}")
return True
except Exception as e:
# Log an error if the plugin fails to load
logging.error(f"Error loading plugin {plugin_name}: {e}")
return False
async def unload_plugin(plugin_name):
"""
Asynchronously unloads a plugin.
Args:
plugin_name (str): The name of the plugin to unload.
Returns:
bool: True if the plugin is unloaded successfully, False otherwise.
"""
try:
if plugin_name in PLUGINS:
del PLUGINS[plugin_name] # Remove the plugin from the PLUGINS dictionary
del sys.modules[f"plugins.{plugin_name}"] # Unload the plugin module from sys.modules
logging.info(f"Unloaded plugin: {plugin_name}")
return True
else:
logging.warning(f"Plugin '{plugin_name}' is not loaded")
return False
except Exception as e:
# Log an error if the plugin fails to unload
logging.error(f"Error unloading plugin {plugin_name}: {e}")
return False
async def handle_command(room, message, bot, prefix, config):
"""
Asynchronously handles the command to load or unload a plugin.
Args:
room (Room): The Matrix room where the command was invoked.
message (RoomMessage): The message object containing the command.
bot (MatrixBot): The Matrix bot instance.
prefix (str): The command prefix.
config (dict): The bot's configuration.
Returns:
None
"""
match = botlib.MessageMatch(room, message, bot, prefix)
if match.is_not_from_this_bot() and match.prefix():
command = match.command()
if command == "load":
if str(message.sender) == config.admin_user:
args = match.args()
if len(args) != 1:
# Send usage message if the command format is incorrect
await bot.api.send_text_message(room.room_id, "Usage: !load <plugin>")
else:
plugin_name = args[0]
# Check if the plugin is not already loaded
if plugin_name not in PLUGINS:
# Load the plugin
success = await load_plugin(plugin_name)
if success:
await bot.api.send_text_message(room.room_id, f"Plugin '{plugin_name}' loaded successfully")
else:
await bot.api.send_text_message(room.room_id, f"Error loading plugin '{plugin_name}'")
else:
await bot.api.send_text_message(room.room_id, f"Plugin '{plugin_name}' is already loaded")
else:
# Send unauthorized message if the sender is not the admin
await bot.api.send_text_message(room.room_id, "You are not authorized to load plugins.")
elif command == "unload":
if str(message.sender) == config.admin_user:
args = match.args()
if len(args) != 1:
# Send usage message if the command format is incorrect
await bot.api.send_text_message(room.room_id, "Usage: !unload <plugin>")
else:
plugin_name = args[0]
# Unload the plugin
success = await unload_plugin(plugin_name)
if success:
await bot.api.send_text_message(room.room_id, f"Plugin '{plugin_name}' unloaded successfully")
else:
await bot.api.send_text_message(room.room_id, f"Error unloading plugin '{plugin_name}'")
else:
# Send unauthorized message if the sender is not the admin
await bot.api.send_text_message(room.room_id, "You are not authorized to unload plugins.")