199 lines
7.7 KiB
Python
199 lines
7.7 KiB
Python
import importlib.util
|
|
import sys
|
|
import os
|
|
import re
|
|
import json
|
|
import stat
|
|
import logging
|
|
import threading
|
|
from xtendr.xtendrbase import XtendRBase
|
|
|
|
__version__ = "0.4.0"
|
|
|
|
logger = logging.getLogger("xtendr")
|
|
|
|
# Plugin (folder) names and module names are restricted to a safe identifier
|
|
# pattern. This blocks path traversal (e.g. "../../etc") and stray characters
|
|
# that have no business in a plugin name.
|
|
_NAME_RE = re.compile(r"^[A-Za-z0-9_-]+$")
|
|
_MODULE_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
|
|
|
|
|
|
class XtendRSystem:
|
|
"""Plugin system to manage plugins.
|
|
|
|
SECURITY NOTE: plugins are arbitrary Python code that runs with the full
|
|
privileges of the host process. Only attach plugins from sources you
|
|
trust. This class validates names/paths and isolates module loading, but
|
|
it cannot make untrusted plugin code safe to run.
|
|
|
|
Example:
|
|
>>> system = XtendRSystem()
|
|
>>> system.version()
|
|
XtendR v0.4.0
|
|
>>> system.attach("example_plugin", lambda: None)
|
|
>>> system.run("example_plugin")
|
|
ExamplePlugin is running!
|
|
>>> system.stop("example_plugin")
|
|
ExamplePlugin has stopped!
|
|
>>> system.detach("example_plugin")
|
|
Detached plugin 'example_plugin'.
|
|
"""
|
|
|
|
def __init__(self, pluginpath="plugins"):
|
|
self.pluginspath = pluginpath
|
|
self.plugins = {}
|
|
self._lock = threading.RLock()
|
|
|
|
def version(self) -> str:
|
|
return "XtendR v" + __version__
|
|
|
|
def _validate_name(self, name: str) -> bool:
|
|
if not isinstance(name, str) or not _NAME_RE.match(name):
|
|
logger.error("Rejected plugin name %r: must match %s", name, _NAME_RE.pattern)
|
|
return False
|
|
return True
|
|
|
|
def _check_permissions(self, path: str) -> None:
|
|
"""Warn (don't block) if a plugin file is group/world-writable."""
|
|
try:
|
|
st = os.stat(path)
|
|
if st.st_mode & (stat.S_IWGRP | stat.S_IWOTH):
|
|
logger.warning(
|
|
"Plugin file '%s' is group/world-writable; this is a "
|
|
"security risk on shared systems.",
|
|
path,
|
|
)
|
|
except OSError:
|
|
pass
|
|
|
|
def attach(self, name: str, callback=None) -> None:
|
|
"""Dynamically load a plugin from its folder."""
|
|
with self._lock:
|
|
if name in self.plugins:
|
|
logger.info("Plugin '%s' is already attached.", name)
|
|
return
|
|
|
|
if not self._validate_name(name):
|
|
return
|
|
|
|
plugin_path = os.path.join(os.getcwd(), self.pluginspath, name)
|
|
info_path = os.path.join(plugin_path, name + ".json")
|
|
|
|
# Defense in depth: even with a validated name, make sure the
|
|
# resolved path is actually inside the plugins directory.
|
|
plugins_root = os.path.realpath(os.path.join(os.getcwd(), self.pluginspath))
|
|
if os.path.commonpath([plugins_root, os.path.realpath(plugin_path)]) != plugins_root:
|
|
logger.error("Refusing to attach '%s': resolves outside plugins directory.", name)
|
|
return
|
|
|
|
if not os.path.isdir(plugin_path) or not os.path.isfile(info_path):
|
|
logger.error("Failed to attach plugin '%s': folder or info file not found.", name)
|
|
return
|
|
|
|
self._check_permissions(info_path)
|
|
|
|
try:
|
|
with open(info_path, "r", encoding="utf-8") as f:
|
|
plugin_info = json.load(f)
|
|
except (OSError, json.JSONDecodeError) as e:
|
|
logger.error("Failed to read info file for plugin '%s': %s", name, e)
|
|
return
|
|
|
|
module_name = plugin_info.get("module")
|
|
class_name = plugin_info.get("class")
|
|
if not module_name or not class_name:
|
|
logger.error("Plugin '%s' info file is missing 'module' or 'class' key.", name)
|
|
return
|
|
if not _MODULE_RE.match(module_name) or not _MODULE_RE.match(class_name):
|
|
logger.error("Plugin '%s' has an invalid module/class identifier.", name)
|
|
return
|
|
|
|
module_file = os.path.join(plugin_path, module_name + ".py")
|
|
if not os.path.isfile(module_file):
|
|
logger.error("Plugin '%s' module file '%s' not found.", name, module_file)
|
|
return
|
|
|
|
self._check_permissions(module_file)
|
|
|
|
# Load the module directly from its file path instead of
|
|
# mutating sys.path. This prevents a plugin from shadowing
|
|
# stdlib or third-party modules for the rest of the process.
|
|
qualified_name = f"xtendr_plugin_{name}_{module_name}"
|
|
try:
|
|
spec = importlib.util.spec_from_file_location(qualified_name, module_file)
|
|
if spec is None or spec.loader is None:
|
|
raise ImportError(f"Could not create import spec for '{module_file}'")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[qualified_name] = module
|
|
spec.loader.exec_module(module)
|
|
|
|
plugin_class = getattr(module, class_name)
|
|
instance = plugin_class()
|
|
|
|
if not isinstance(instance, XtendRBase):
|
|
logger.error("Plugin '%s' does not inherit from XtendRBase.", name)
|
|
sys.modules.pop(qualified_name, None)
|
|
return
|
|
|
|
except Exception as e: # noqa: BLE001 - plugin code is untrusted, isolate all failures
|
|
logger.error("Failed to attach plugin '%s': %s", name, e, exc_info=True)
|
|
sys.modules.pop(qualified_name, None)
|
|
return
|
|
|
|
self.plugins[name] = {
|
|
"instance": instance,
|
|
"running": False,
|
|
"info": plugin_info,
|
|
"autorun": False,
|
|
"module_key": qualified_name,
|
|
}
|
|
logger.info("Attached plugin '%s'.", name)
|
|
logger.info("Running pre-load on '%s'.", name)
|
|
|
|
def _pre_load_worker():
|
|
try:
|
|
instance.pre_load(callback)
|
|
except Exception: # noqa: BLE001
|
|
logger.error("Plugin '%s' raised during pre_load.", name, exc_info=True)
|
|
|
|
thread = threading.Thread(target=_pre_load_worker, daemon=True)
|
|
thread.start()
|
|
|
|
def run(self, name: str, *args, **kwargs):
|
|
"""Run the plugin's 'run' method if available."""
|
|
with self._lock:
|
|
entry = self.plugins.get(name)
|
|
if entry is None:
|
|
logger.error("Plugin '%s' not found or has no 'run' method.", name)
|
|
return
|
|
entry["running"] = True
|
|
try:
|
|
return entry["instance"].run(*args, **kwargs)
|
|
except Exception: # noqa: BLE001
|
|
logger.error("Plugin '%s' raised during run.", name, exc_info=True)
|
|
entry["running"] = False
|
|
|
|
def stop(self, name: str) -> None:
|
|
"""Stop the plugin if it's running."""
|
|
with self._lock:
|
|
entry = self.plugins.get(name)
|
|
if entry is None or not entry["running"]:
|
|
logger.info("Plugin '%s' is not running.", name)
|
|
return
|
|
entry["running"] = False
|
|
try:
|
|
entry["instance"].stop()
|
|
except Exception: # noqa: BLE001
|
|
logger.error("Plugin '%s' raised during stop.", name, exc_info=True)
|
|
|
|
def detach(self, name: str) -> None:
|
|
"""Unload a plugin."""
|
|
with self._lock:
|
|
entry = self.plugins.pop(name, None)
|
|
if entry is None:
|
|
logger.info("Plugin '%s' is not attached.", name)
|
|
return
|
|
sys.modules.pop(entry["module_key"], None)
|
|
logger.info("Detached plugin '%s'.", name)
|