hash/FunguyBot
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

various plugin refactors and fixes

Browse Source
This commit is contained in:
Hash Borgir
2026-05-09 12:08:38 -05:00
parent 5c6234a317
commit b722a78d21
6 changed files with 402 additions and 334 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+21 -17
View File
@@ -104,16 +104,18 @@ systemctl start funguybot
The bot includes the following plugins:
- **admin.py**: Full room moderation – multi‑word name support
- **arxiv.py**: arXiv academic paper search (with rate limiting and error reporting)
- **admin.py**: Full room moderation – multi-word name support
- **arxiv.py**: arXiv academic paper search
- **bitcoin.py**: Current Bitcoin price
- **common.py**: Shared utilities for FunguyBot plugins.
- **config.py**: Admin-only configuration commands (preserves disabled plugins)
- **cron.py**: In‑process cron scheduler (room‑aware, no system crontab)
- **cron.py**: In-process cron scheduler (room-aware, no system crontab)
- **date.py**: Show current date and time
- **ddg.py**: DuckDuckGo search – collapsible results (ddgs library, no API key)
- **dictionary.py** Dictionary plugin to fetch word definitions from dict
- **dns.py**: DNS reconnaissance
- **ddg.py**: DuckDuckGo search plugin
- **dictionary.py**: Look up word definitions using system dictionary
- **dns.py**: DNS reconnaissance (SSRF-safe)
- **dnsdumpster.py**: DNSDumpster domain reconnaissance
- **encode.py**: Comprehensive CyberChef-like encoding and analysis toolkit
- **exploitdb.py**: Exploit-DB search
- **fortune.py**: Random fortune message
- **geo.py**: IP geolocation lookup
@@ -124,27 +126,29 @@ The bot includes the following plugins:
- **imdb.py**: IMDb lookup via OMDb API
- **infermatic-text.py**: AI text generation via Infermatic API
- **isup.py**: Check if a site is up
- **joke.py**: Get a random joke from the joke APIs
- **joke.py**: Get random jokes from the Official Joke API
- **karma.py**: Room karma tracking system (display names only, no Matrix IDs)
- **lastfm.py**: Last.fm integration
- **lastfm.py**: Last.fm music stats with aligned code block output
- **loadplugin.py**: Load/unload plugins at runtime
- **news.py**: News headlines via GNews API
- **plugins.py**: List all loaded plugins
- **plugins.py**: List all loaded plugins with count
- **proxy.py**: Working SOCKS5 proxy finder
- **quote.py**: Goodreads quotes via headless browser (Playwright)
- **roomstats.py**: Per‑user room statistics (Limnoria‑style), with multi‑word name support
- **quote.py**: Fetch Goodreads quotes
- **roomstats.py**: Per-user room statistics
- **shodan.py**: Shodan.io reconnaissance
- **sslscan.py**: SSL/TLS security scanner
- **stable-diffusion.py**: Stable Diffusion image generation
- **stable-diffusion.py**: Stable Diffusion image generation (LORA support)
- **subdomains.py**: Subdomain enumeration via CertSpotter
- **sysinfo.py**: System information and monitoring
- **timezone.py**: World clock (no hardcoded cities)
- **subnet.py**: Subnet calculator
- **sysinfo.py**: System information plugin
- **timezone.py**: World clock (offline IANA zones + free geocoding)
- **urbandictionary.py**: Urban Dictionary definitions
- **weather.py**: Weather forecast (OWM primary, Open‑Meteo fallback)
- **utils.py**: Security utilities for Funguy Bot plugins.
- **weather.py**: Weather data plugin
- **welcome.py**: Room welcome message
- **whois.py**: WHOIS lookup
- **whois.py**: Domain WHOIS lookup
- **wikipedia.py**: Wikipedia article summary
- **xkcd.py**: Random XKCD comic
- **xkcd.py**: Fetch random or specific xkcd comics
- **youtube-search.py**: YouTube video search
## Configuration
funguy.py
+122 -34
View File
@@ -20,7 +20,7 @@ from plugins.config import FunguyConfig
# Rate limiter settings
RATE_LIMIT_WINDOW = 5.0 # seconds
MAX_COMMANDS_PER_WINDOW = 5
MAX_COMMANDS_PER_WINDOW = 3
class FunguyBot:
@@ -118,22 +118,78 @@ class FunguyBot:
toml.dump(existing_config, f)
def _check_rate_limit(self, sender: str) -> bool:
"""Return True if the sender is allowed to proceed, False if rate limited."""
"""Return True if the sender is allowed to proceed.
Admin is always allowed."""
# Admin bypass
if sender == self.config.admin_user:
return True
now = time.monotonic()
bucket = self._rate_limit_buckets[sender]
# Prune old entries
bucket = [t for t in bucket if now - t < RATE_LIMIT_WINDOW]
self._rate_limit_buckets[sender] = bucket
if len(bucket) >= MAX_COMMANDS_PER_WINDOW:
logging.debug("Rate limit hit for %s", sender)
return False
bucket.append(now)
return True
# ------------------------------------------------------------------
# New: load/unload a single plugin at runtime
# ------------------------------------------------------------------
async def load_plugin(self, plugin_name: str) -> bool:
"""Dynamically load a plugin module, add to PLUGINS, and call its setup()."""
if plugin_name in self.PLUGINS:
logging.info(f"Plugin '{plugin_name}' is already loaded.")
return False
try:
module = importlib.import_module(f"{self.PLUGINS_DIR}.{plugin_name}")
self.PLUGINS[plugin_name] = module
logging.info(f"Loaded plugin: {plugin_name}")
# Call setup if the bot is already running
if self.bot is not None and hasattr(module, "setup") and callable(module.setup):
module.setup(self.bot)
logging.info(f"Setup called for newly loaded plugin: {plugin_name}")
return True
except Exception as e:
logging.error(f"Error loading plugin {plugin_name}: {e}")
return False
async def unload_plugin(self, plugin_name: str) -> bool:
"""Remove a plugin from PLUGINS and unload its module."""
if plugin_name not in self.PLUGINS:
logging.info(f"Plugin '{plugin_name}' is not loaded.")
return False
try:
del self.PLUGINS[plugin_name]
module_path = f"{self.PLUGINS_DIR}.{plugin_name}"
if module_path in sys.modules:
del sys.modules[module_path]
logging.info(f"Unloaded plugin: {plugin_name}")
return True
except Exception as e:
logging.error(f"Error unloading plugin {plugin_name}: {e}")
return False
# ------------------------------------------------------------------
# New: restart the bot process
# ------------------------------------------------------------------
async def restart_bot(self, room_id):
await self.bot.api.send_text_message(room_id, "🔄 Restarting bot...")
await asyncio.sleep(1)
logging.info("Restart command received – exiting.")
sys.exit(0)
async def handle_commands(self, room, message):
match = botlib.MessageMatch(room, message, self.bot, self.config.prefix)
# Rate limit check (applies to all commands)
sender = str(message.sender)
is_admin = (sender == self.config.admin_user)
# Rate limit check (applies to all non‑admin commands)
if not self._check_rate_limit(sender):
await self.bot.api.send_text_message(
room.room_id,
@@ -143,45 +199,81 @@ class FunguyBot:
# Admin commands
if match.is_not_from_this_bot() and match.prefix() and match.command("reload"):
if sender == self.config.admin_user:
if is_admin:
self.reload_plugins()
await self.bot.api.send_text_message(room.room_id, "Plugins reloaded successfully")
await self.bot.api.send_text_message(room.room_id, "All plugins reloaded successfully.")
else:
await self.bot.api.send_text_message(room.room_id, "You are not authorized to reload plugins.")
return
if match.is_not_from_this_bot() and match.prefix() and match.command("load"):
if not is_admin:
await self.bot.api.send_text_message(room.room_id, "⛔ Admin only.")
return
args = match.args()
if len(args) != 1:
await self.bot.api.send_text_message(room.room_id, "Usage: !load <plugin>")
return
success = await self.load_plugin(args[0])
msg = f"✅ Plugin '{args[0]}' loaded." if success else f"❌ Could not load '{args[0]}'. See logs for details."
await self.bot.api.send_text_message(room.room_id, msg)
return
if match.is_not_from_this_bot() and match.prefix() and match.command("unload"):
if not is_admin:
await self.bot.api.send_text_message(room.room_id, "⛔ Admin only.")
return
args = match.args()
if len(args) != 1:
await self.bot.api.send_text_message(room.room_id, "Usage: !unload <plugin>")
return
success = await self.unload_plugin(args[0])
msg = f"✅ Plugin '{args[0]}' unloaded." if success else f"❌ Could not unload '{args[0]}'. See logs for details."
await self.bot.api.send_text_message(room.room_id, msg)
return
if match.is_not_from_this_bot() and match.prefix() and match.command("disable"):
if sender == self.config.admin_user:
args = match.args()
if len(args) != 2:
await self.bot.api.send_text_message(room.room_id, "Usage: !disable <plugin> <room_id>")
else:
plugin_name, room_id = args
await self.disable_plugin(room_id, plugin_name)
await self.bot.api.send_text_message(room.room_id, f"Plugin '{plugin_name}' disabled for room '{room_id}'")
else:
await self.bot.api.send_text_message(room.room_id, "You are not authorized to disable plugins.")
if not is_admin:
await self.bot.api.send_text_message(room.room_id, "⛔ Admin only.")
return
args = match.args()
if len(args) != 1:
await self.bot.api.send_text_message(room.room_id, "Usage: !disable <plugin>")
return
plugin_name = args[0]
room_id = room.room_id
await self.disable_plugin(room_id, plugin_name)
await self.bot.api.send_text_message(room.room_id, f"🚫 Plugin '{plugin_name}' disabled in this room.")
return
if match.is_not_from_this_bot() and match.prefix() and match.command("enable"):
if sender == self.config.admin_user:
args = match.args()
if len(args) != 2:
await self.bot.api.send_text_message(room.room_id, "Usage: !enable <plugin> <room_id>")
else:
plugin_name, room_id = args
await self.enable_plugin(room_id, plugin_name)
await self.bot.api.send_text_message(room.room_id, f"Plugin '{plugin_name}' enabled for room '{room_id}'")
else:
await self.bot.api.send_text_message(room.room_id, "You are not authorized to enable plugins.")
if not is_admin:
await self.bot.api.send_text_message(room.room_id, "⛔ Admin only.")
return
args = match.args()
if len(args) != 1:
await self.bot.api.send_text_message(room.room_id, "Usage: !enable <plugin>")
return
plugin_name = args[0]
room_id = room.room_id
await self.enable_plugin(room_id, plugin_name)
await self.bot.api.send_text_message(room.room_id, f"✅ Plugin '{plugin_name}' enabled in this room.")
return
if match.is_not_from_this_bot() and match.prefix() and match.command("restart"):
if not is_admin:
await self.bot.api.send_text_message(room.room_id, "⛔ Admin only.")
return
await self.restart_bot(room.room_id)
return
if match.is_not_from_this_bot() and match.prefix() and match.command("rehash"):
if sender == self.config.admin_user:
self.rehash_config()
await self.bot.api.send_text_message(room.room_id, "Config rehashed")
else:
await self.bot.api.send_text_message(room.room_id, "You are not authorized to reload plugins.")
if not is_admin:
await self.bot.api.send_text_message(room.room_id, "You are not authorized to reload config.")
return
self.load_config()
self.load_disabled_plugins()
await self.bot.api.send_text_message(room.room_id, "🔄 Configuration rehashed.")
return
# Dispatch to active plugins
@@ -192,10 +284,6 @@ class FunguyBot:
except Exception as e:
logging.error(f"Error in plugin {plugin_name}: {e}", exc_info=True)
def rehash_config(self):
del self.config
self.config = FunguyConfig()
async def disable_plugin(self, room_id, plugin_name):
if room_id not in self.disabled_plugins:
self.disabled_plugins[room_id] = []
plugins/config.py
+258 -94
View File
@@ -2,6 +2,14 @@
Custom configuration class for the Funguy bot.
Security‑hardened: only the configured admin user can read or change settings.
Save operation preserves extra sections (plugins.disabled, etc.).
Commands:
!set <option> <value>
!get <option>
!show
!saveconf
!loadconf
!reset
!config help
"""
# plugins/config.py
@@ -10,6 +18,75 @@ import logging
import toml
import simplematrixbotlib as botlib
from dataclasses import dataclass
from plugins.common import code_block, collapsible_summary, html_escape
# ----------------------------------------------------------------------
# Allowed configuration keys and metadata
# ----------------------------------------------------------------------
OPTIONS = {
"prefix": {
"description": "Command prefix (e.g., !)",
"validator": lambda v: isinstance(v, str) and len(v) == 1,
"type": "string",
"default": "!",
},
"timeout": {
"description": "HTTP timeout (seconds)",
"validator": lambda v: isinstance(v, (int, float)) and v > 0,
"type": "integer",
"default": 30,
},
"join_on_invite": {
"description": "Auto‑join rooms on invite (true/false)",
"validator": lambda v: isinstance(v, bool),
"type": "boolean",
"default": True,
},
"encryption_enabled": {
"description": "Enable message encryption (true/false)",
"validator": lambda v: isinstance(v, bool),
"type": "boolean",
"default": False,
},
"emoji_verify": {
"description": "Use emoji verification (true/false)",
"validator": lambda v: isinstance(v, bool),
"type": "boolean",
"default": False,
},
"ignore_unverified_devices": {
"description": "Ignore unverified devices (true/false)",
"validator": lambda v: isinstance(v, bool),
"type": "boolean",
"default": True,
},
"store_path": {
"description": "Path for device store",
"validator": lambda v: isinstance(v, str),
"type": "string",
"default": "./store/",
},
"allowlist": {
"description": "Allowed users (comma separated)",
"validator": lambda v: isinstance(v, list),
"type": "list",
"default": [],
},
"blocklist": {
"description": "Blocked users (comma separated)",
"validator": lambda v: isinstance(v, list),
"type": "list",
"default": [],
},
"admin_user": {
"description": "Admin Matrix user ID",
"validator": lambda v: isinstance(v, str) and v.startswith("@"),
"type": "string (read‑only via !set)",
"default": "", # must be set via config file
"readonly": True,
},
}
@dataclass
@@ -17,16 +94,24 @@ class FunguyConfig(botlib.Config):
"""
Custom configuration class for the Funguy bot.
Extends the base Config class to provide additional configuration options.
Args:
config_file (str): Path to the configuration file.
"""
def __init__(self, config_file="funguy.conf"):
super().__init__()
# Load configuration from file
# Load the TOML file *first* so we can easily extract admin_user and prefix
raw = {}
if os.path.exists(config_file):
with open(config_file, 'r') as f:
raw = toml.load(f)
bot_section = raw.get('simplematrixbotlib', {}).get('config', {})
self._admin_user = bot_section.get('admin_user', '')
self._prefix = bot_section.get('prefix', '!')
# Now let the base class parse the file and fill in other attributes
self.load_toml(config_file)
# Store the actual file path used (so save_config can find it)
# Store the file path for later saves
self._config_file = config_file
logging.info(f"Loaded configuration from {config_file}")
@@ -59,7 +144,16 @@ class FunguyConfig(botlib.Config):
self._config_file = value
def load_config(self, config_file):
"""Load configuration from a TOML file."""
"""Load configuration from a TOML file (re‑read admin_user and prefix)."""
raw = {}
if os.path.exists(config_file):
with open(config_file, 'r') as f:
raw = toml.load(f)
bot_section = raw.get('simplematrixbotlib', {}).get('config', {})
self._admin_user = bot_section.get('admin_user', '')
self._prefix = bot_section.get('prefix', '!')
self.load_toml(config_file)
self._config_file = config_file
logging.info(f"Loaded configuration from {config_file}")
@@ -68,37 +162,28 @@ class FunguyConfig(botlib.Config):
"""
Save configuration to a TOML file, **preserving** any extra sections
(e.g., plugins.disabled) not managed by the base library.
If config_file is not provided, the instance's stored config_file is used.
"""
if config_file is None:
config_file = self._config_file
if not config_file:
raise ValueError("No config file path set for saving.")
# 1. Let the library write its portion to a temporary file
tmp_file = config_file + ".tmp"
try:
self.save_toml(tmp_file)
# 2. Read the temporary file (library's view of the config)
with open(tmp_file, 'r') as f:
new_config = toml.load(f)
# 3. Read the current config file (if it exists) to preserve extra sections
original = {}
if os.path.exists(config_file):
with open(config_file, 'r') as f:
original = toml.load(f)
# 4. Merge: keep everything from the original, then overlay
# the library's config table(s). This leaves any top-level
# sections not produced by the library exactly as they were.
merged = original.copy()
for key, value in new_config.items():
merged[key] = value
# 5. Write back the merged result
with open(config_file, 'w') as f:
toml.dump(merged, f)
@@ -107,31 +192,95 @@ class FunguyConfig(botlib.Config):
logging.error(f"Error saving config to {config_file}: {e}")
raise
finally:
# Always remove the temp file
if os.path.exists(tmp_file):
os.remove(tmp_file)
async def handle_command(room, message, bot, prefix, config):
"""
Handle commands related to bot configuration.
All sub‑commands require the sender to be the configured admin_user.
"""
match = botlib.MessageMatch(room, message, bot, prefix)
# ----------------------------------------------------------------------
# Helpers for formatting
# ----------------------------------------------------------------------
def _bool_to_str(v):
return "true" if v else "false"
if not match.is_not_from_this_bot() or not match.prefix():
def _str_to_bool(s: str):
return s.lower() in ("true", "yes", "1")
def _format_value(key, value):
if isinstance(value, bool):
return _bool_to_str(value)
if isinstance(value, list):
return ", ".join(value) if value else "(empty)"
return str(value)
def _set_config_option(config, key, value_str):
meta = OPTIONS.get(key)
if not meta:
return False, f"Unknown option '{html_escape(key)}'."
if meta.get("readonly"):
return False, f"'{html_escape(key)}' cannot be changed via !set for safety."
try:
if meta["type"] == "boolean":
value = _str_to_bool(value_str)
elif meta["type"] == "integer":
value = int(value_str)
elif meta["type"] == "list":
if value_str.strip() == "":
value = []
else:
value = [item.strip() for item in value_str.split(",") if item.strip()]
else:
value = value_str
if not meta["validator"](value):
return False, f"Invalid value for '{html_escape(key)}'. Expected {meta['type']}."
# Apply to config object
if key == "prefix":
config.prefix = value
elif key == "timeout":
config.timeout = value
elif key == "join_on_invite":
config.join_on_invite = value
elif key == "encryption_enabled":
config.encryption_enabled = value
elif key == "emoji_verify":
config.emoji_verify = value
elif key == "ignore_unverified_devices":
config.ignore_unverified_devices = value
elif key == "store_path":
config.store_path = value
elif key == "allowlist":
config.allowlist = value
elif key == "blocklist":
config.blocklist = value
# admin_user is readonly, not set here
return True, f"✅ {html_escape(key)} set to {_format_value(key, value)}."
except (ValueError, TypeError) as e:
return False, f"Invalid value: {html_escape(str(e))}"
# ----------------------------------------------------------------------
# Command handler
# ----------------------------------------------------------------------
async def handle_command(room, message, bot, prefix, config):
match = botlib.MessageMatch(room, message, bot, prefix)
if not (match.is_not_from_this_bot() and match.prefix()):
return
cmd = match.command()
if cmd not in ("set", "get", "saveconf", "loadconf", "show", "reset"):
if cmd not in ("config", "set", "get", "show", "saveconf", "loadconf", "reset"):
return
sender = str(message.sender)
if sender != config.admin_user:
logging.warning(
"Unauthorized config command attempt by %s in room %s: %s",
sender, room.room_id, cmd
)
logging.warning("Unauthorized config command attempt by %s (%s vs %s)", sender, config.admin_user)
await bot.api.send_text_message(
room.room_id,
"⛔ You are not authorized to use configuration commands."
@@ -139,94 +288,109 @@ async def handle_command(room, message, bot, prefix, config):
return
args = match.args()
if cmd == "config":
if not args:
await _send_help(room, bot)
return
subcmd = args[0].lower()
args = args[1:]
else:
subcmd = cmd
if cmd == "set":
if subcmd == "set":
if len(args) != 2:
await bot.api.send_text_message(room.room_id,
"Usage: !set <config_option> <value>")
await bot.api.send_text_message(room.room_id, "Usage: !set <option> <value>\nUse !config show for options.")
return
option, value = args
if option == "admin_user":
await bot.api.send_text_message(
room.room_id,
"❌ Changing 'admin_user' via !set is not allowed for security reasons."
)
return
elif option == "prefix":
config.prefix = value
await bot.api.send_text_message(room.room_id,
f"Prefix set to `{value}`")
else:
await bot.api.send_text_message(room.room_id,
"Invalid configuration option.")
success, msg = _set_config_option(config, option, value)
await bot.api.send_text_message(room.room_id, msg)
elif cmd == "get":
elif subcmd == "get":
if len(args) != 1:
await bot.api.send_text_message(room.room_id,
"Usage: !get <config_option>")
await bot.api.send_text_message(room.room_id, "Usage: !get <option>")
return
option = args[0]
if option == "admin_user":
await bot.api.send_text_message(room.room_id,
f"Admin user: {config.admin_user}")
elif option == "prefix":
await bot.api.send_text_message(room.room_id,
f"Prefix: {config.prefix}")
else:
await bot.api.send_text_message(room.room_id,
"Invalid configuration option.")
meta = OPTIONS.get(option)
if not meta:
await bot.api.send_text_message(room.room_id, f"Unknown option '{html_escape(option)}'.")
return
val = getattr(config, option, meta["default"])
await bot.api.send_text_message(room.room_id, f"{html_escape(option)}: {_format_value(option, val)}")
elif cmd == "show":
await bot.api.send_text_message(
room.room_id,
f"Admin user: {config.admin_user}\nPrefix: {config.prefix}"
)
elif subcmd == "show":
rows = []
for key, meta in OPTIONS.items():
val = getattr(config, key, meta["default"])
rows.append(("⚙️", key, _format_value(key, val)))
block = code_block("📋 Current Configuration", [{"title": "", "rows": rows}])
output = collapsible_summary("📋 Current Configuration", block)
await bot.api.send_markdown_message(room.room_id, output)
elif cmd == "saveconf":
elif subcmd == "saveconf":
try:
config.save_config() # uses the stored config_file by default
await bot.api.send_text_message(
room.room_id,
"Configuration saved (including disabled plugins)."
)
config.save_config()
await bot.api.send_text_message(room.room_id, "💾 Configuration saved to file.")
except Exception as e:
await bot.api.send_text_message(
room.room_id,
f"❌ Failed to save configuration: {e}"
)
await bot.api.send_text_message(room.room_id, f"❌ Failed to save: {html_escape(str(e))}")
elif cmd == "loadconf":
config.load_config(config.config_file)
await bot.api.send_text_message(
room.room_id,
"Configuration reloaded from file."
)
elif subcmd == "loadconf":
try:
config.load_config(config.config_file)
await bot.api.send_text_message(room.room_id, "🔄 Configuration reloaded from file.")
except Exception as e:
await bot.api.send_text_message(room.room_id, f"❌ Failed to load: {html_escape(str(e))}")
elif cmd == "reset":
config.prefix = "!"
await bot.api.send_text_message(
room.room_id,
"Configuration reset to defaults (admin_user unchanged)."
)
elif subcmd == "reset":
for key, meta in OPTIONS.items():
if key == "admin_user":
continue
setattr(config, key, meta["default"])
if key == "prefix":
config.prefix = meta["default"]
await bot.api.send_text_message(room.room_id, "♻️ Configuration reset to defaults (admin_user preserved).")
elif subcmd == "help":
await _send_help(room, bot)
else:
await bot.api.send_text_message(room.room_id, f"Unknown subcommand '{html_escape(subcmd)}'. Use !config help.")
async def _send_help(room, bot):
help_text = """
<details>
<summary><strong>🔧 Config Plugin Commands</strong></summary>
<p><code>!set &lt;option&gt; &lt;value&gt;</code> – Change a configuration option</p>
<p><code>!get &lt;option&gt;</code> – Display a single option</p>
<p><code>!show</code> – Show all current settings</p>
<p><code>!saveconf</code> – Save configuration to file</p>
<p><code>!loadconf</code> – Reload from file</p>
<p><code>!reset</code> – Reset to defaults (keeps admin_user)</p>
<p><code>!config help</code> – This help</p>
<p><strong>Available options:</strong><br>
<code>prefix, timeout, join_on_invite, encryption_enabled, emoji_verify, ignore_unverified_devices, store_path, allowlist, blocklist</code></p>
<p><em>Note: <code>admin_user</code> can only be changed by editing funguy.conf directly.</em></p>
</details>
"""
await bot.api.send_markdown_message(room.room_id, help_text)
# ---------------------------------------------------------------------------
# Plugin Metadata
# ---------------------------------------------------------------------------
__version__ = "1.0.2"
__author__ = "Funguy Bot (hardened)"
__description__ = "Admin-only configuration commands (preserves disabled plugins)"
__version__ = "1.1.1"
__author__ = "Funguy Bot"
__description__ = "Admin‑only configuration management"
__help__ = """
<details>
<summary><strong>Admin Config</strong> (!set, !get, !saveconf, …)</summary>
<summary><strong>!config</strong> – Manage bot settings</summary>
<ul>
<li><code>!set prefix &lt;value&gt;</code> – Change command prefix (admin only)</li>
<li><code>!get &lt;option&gt;</code> – Display config value (admin only)</li>
<li><code>!show</code> – Show current settings (admin only)</li>
<li><code>!saveconf</code> / <code>!loadconf</code> – Save/load config (admin only)</li>
<li><code>!reset</code> – Reset to defaults, preserving admin_user (admin only)</li>
<li><code>!set &lt;option&gt; &lt;value&gt;</code> – Change a setting</li>
<li><code>!get &lt;option&gt;</code> – View a setting</li>
<li><code>!show</code> – All settings</li>
<li><code>!saveconf</code> – Save to file</li>
<li><code>!loadconf</code> – Reload from file</li>
<li><code>!reset</code> – Reset defaults</li>
<li><code>!config help</code> – This help</li>
</ul>
<p>Changing <code>admin_user</code> via bot commands is blocked for safety.</p>
<p>The <code>plugins.disabled</code> section is now preserved when saving.</p>
</details>
"""
plugins/isup.py
+1 -2
View File
@@ -6,8 +6,7 @@ import logging
import aiohttp
import socket
import simplematrixbotlib as botlib
from plugins.utils import is_public_destination
from plugins.common import is_public_destination
async def check_http(domain):
"""Check if HTTP service is up for the given domain."""
plugins/loadplugin.py
-128
View File
@@ -1,128 +0,0 @@
"""
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 = {}
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.
"""
try:
# Import the plugin module
module = importlib.import_module(f"plugins.{plugin_name}")
# 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.")
# ---------------------------------------------------------------------------
# Plugin Metadata
# ---------------------------------------------------------------------------
__version__ = "1.0.0"
__author__ = "Funguy Bot"
__description__ = "Load/unload plugins at runtime"
__help__ = """
<details>
<summary><strong>Admin: !load / !unload</strong></summary>
<p><code>!load &lt;plugin&gt;</code> / <code>!unload &lt;plugin&gt;</code> – Dynamically load or unload a plugin module. Admin only.</p>
</details>
"""
plugins/utils.py
-59
View File
@@ -1,59 +0,0 @@
"""
Security utilities for Funguy Bot plugins.
"""
import ipaddress
import socket
import logging
logger = logging.getLogger("security_utils")
# Networks considered unsafe for outbound connections
PRIVATE_RANGES = [
ipaddress.ip_network('10.0.0.0/8'),
ipaddress.ip_network('172.16.0.0/12'),
ipaddress.ip_network('192.168.0.0/16'),
ipaddress.ip_network('127.0.0.0/8'),
ipaddress.ip_network('169.254.0.0/16'), # link‑local
ipaddress.ip_network('0.0.0.0/8'), # "this" network
ipaddress.ip_network('::1/128'), # IPv6 loopback
ipaddress.ip_network('fc00::/7'), # unique local
ipaddress.ip_network('fe80::/10'), # link‑local
ipaddress.ip_network('::/128'), # unspecified
]
def is_public_destination(target: str) -> bool:
"""
Returns True if `target` (hostname or IP) does NOT resolve to any
private, loopback, or link‑local address.
"""
try:
# Try parsing as an IP address first
addr = ipaddress.ip_address(target)
if any(addr in net for net in PRIVATE_RANGES):
return False
return True
except ValueError:
pass
# Resolve hostname to IPs
try:
addrinfo = socket.getaddrinfo(target, None)
for _, _, _, _, sockaddr in addrinfo:
ip = sockaddr[0]
addr = ipaddress.ip_address(ip)
if any(addr in net for net in PRIVATE_RANGES):
return False
return True
except Exception as e:
logger.warning(f"Cannot resolve {target}: {e}")
return False
# ---------------------------------------------------------------------------
# No‑op command handler – prevents bot crash because funguy.py calls
# handle_command() on every module in the plugins directory.
# ---------------------------------------------------------------------------
async def handle_command(room, message, bot, prefix, config):
"""This module is not a command plugin; ignore all messages."""
pass