397 lines
13 KiB
Python
397 lines
13 KiB
Python
"""
|
||
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
|
||
|
||
import os
|
||
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
|
||
class FunguyConfig(botlib.Config):
|
||
"""
|
||
Custom configuration class for the Funguy bot.
|
||
Extends the base Config class to provide additional configuration options.
|
||
"""
|
||
def __init__(self, config_file="funguy.conf"):
|
||
super().__init__()
|
||
|
||
# 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 file path for later saves
|
||
self._config_file = config_file
|
||
logging.info(f"Loaded configuration from {config_file}")
|
||
|
||
_admin_user: str = ""
|
||
_prefix: str = ""
|
||
_config_file: str= ""
|
||
|
||
@property
|
||
def admin_user(self):
|
||
return self._admin_user
|
||
|
||
@admin_user.setter
|
||
def admin_user(self, value):
|
||
self._admin_user = value
|
||
|
||
@property
|
||
def prefix(self):
|
||
return self._prefix
|
||
|
||
@prefix.setter
|
||
def prefix(self, value):
|
||
self._prefix = value
|
||
|
||
@property
|
||
def config_file(self):
|
||
return self._config_file
|
||
|
||
@config_file.setter
|
||
def config_file(self, value):
|
||
self._config_file = value
|
||
|
||
def load_config(self, config_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}")
|
||
|
||
def save_config(self, config_file=None):
|
||
"""
|
||
Save configuration to a TOML file, **preserving** any extra sections
|
||
(e.g., plugins.disabled) not managed by the base library.
|
||
"""
|
||
if config_file is None:
|
||
config_file = self._config_file
|
||
if not config_file:
|
||
raise ValueError("No config file path set for saving.")
|
||
|
||
tmp_file = config_file + ".tmp"
|
||
try:
|
||
self.save_toml(tmp_file)
|
||
|
||
with open(tmp_file, 'r') as f:
|
||
new_config = toml.load(f)
|
||
|
||
original = {}
|
||
if os.path.exists(config_file):
|
||
with open(config_file, 'r') as f:
|
||
original = toml.load(f)
|
||
|
||
merged = original.copy()
|
||
for key, value in new_config.items():
|
||
merged[key] = value
|
||
|
||
with open(config_file, 'w') as f:
|
||
toml.dump(merged, f)
|
||
|
||
logging.info(f"Configuration saved to {config_file} (extra sections preserved)")
|
||
except Exception as e:
|
||
logging.error(f"Error saving config to {config_file}: {e}")
|
||
raise
|
||
finally:
|
||
if os.path.exists(tmp_file):
|
||
os.remove(tmp_file)
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# Helpers for formatting
|
||
# ----------------------------------------------------------------------
|
||
def _bool_to_str(v):
|
||
return "true" if v else "false"
|
||
|
||
|
||
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 ("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 (%s vs %s)", sender, config.admin_user)
|
||
await bot.api.send_text_message(
|
||
room.room_id,
|
||
"⛔ You are not authorized to use configuration commands."
|
||
)
|
||
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 subcmd == "set":
|
||
if len(args) != 2:
|
||
await bot.api.send_text_message(room.room_id, "Usage: !set <option> <value>\nUse !config show for options.")
|
||
return
|
||
option, value = args
|
||
success, msg = _set_config_option(config, option, value)
|
||
await bot.api.send_text_message(room.room_id, msg)
|
||
|
||
elif subcmd == "get":
|
||
if len(args) != 1:
|
||
await bot.api.send_text_message(room.room_id, "Usage: !get <option>")
|
||
return
|
||
option = args[0]
|
||
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 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 subcmd == "saveconf":
|
||
try:
|
||
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: {html_escape(str(e))}")
|
||
|
||
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 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 <option> <value></code> – Change a configuration option</p>
|
||
<p><code>!get <option></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.1.1"
|
||
__author__ = "Funguy Bot"
|
||
__description__ = "Admin‑only configuration management"
|
||
__help__ = """
|
||
<details>
|
||
<summary><strong>!config</strong> – Manage bot settings</summary>
|
||
<ul>
|
||
<li><code>!set <option> <value></code> – Change a setting</li>
|
||
<li><code>!get <option></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>
|
||
</details>
|
||
"""
|