import importlib.util import sys import os import re import json import stat import logging import threading from pathlib import Path from xtendr.xtendrbase import XtendRBase from xtendr import signing as xsign __version__ = "0.5.1" 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_]*$") # Signature status values stored in plugins[name]["signature"]["status"]. SIG_VERIFIED = "verified" # whitelist entry present, signature and hash both check out SIG_UNSIGNED = "unsigned" # no whitelist entry for this plugin at all SIG_INVALID = "invalid" # whitelist entry present but signature/hash mismatch (tampered) 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.5.1 >>> 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'. A system can be marked `protected=True` for plugins that must stay attached for the lifetime of the process (e.g. an application's built-ins). Protected systems still attach/run/stop normally; only detach() is restricted, and only for ordinary callers -- detach(..., force=True) and detach_all(..., force=True) remain available for the host application's own shutdown path. There is deliberately no way to force a single detach() without also opting in via the same keyword a UI action would have to expose, so "not detachable by the user" is a property of the call site, not of hidden state. """ def __init__(self, pluginpath="plugins", public_key_path=None, whitelist_path=None, whitelist_passphrase=None, protected=False): self.pluginspath = pluginpath self.plugins = {} self.protected = protected self._lock = threading.RLock() # -- signature verification setup --------------------------------- # If either the public key or the whitelist can't be loaded, we # fail closed: self._public_key / self._whitelist stay None, and # every plugin will come back as SIG_UNSIGNED (disabled) rather # than silently skipping verification. This is deliberate -- an # admin who wants unsigned plugins to run should not be able to # get there by accident (e.g. a missing/misspelled key path). self._public_key = None self._whitelist = None if public_key_path is not None: try: self._public_key = xsign.load_public_key(Path(public_key_path)) except (OSError, ValueError) as e: logger.error("Could not load XtendR public key from '%s': %s", public_key_path, e) if whitelist_path is not None: try: self._whitelist = xsign.Whitelist.load(Path(whitelist_path), whitelist_passphrase) except (OSError, ValueError) as e: logger.error("Could not load XtendR plugin whitelist from '%s': %s", whitelist_path, e) if self._public_key is None or self._whitelist is None: logger.warning( "Signature verification is not fully configured for pluginpath '%s'; " "all plugins will be treated as unsigned and permanently disabled.", pluginpath, ) def version(self) -> str: return "XtendR v" + __version__ @property def verification_configured(self) -> bool: """True if a public key and whitelist both loaded successfully.""" return self._public_key is not None and self._whitelist is not None 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 _verify_signature(self, name: str, plugin_path: str, module_name: str) -> dict: """Check a plugin's signature against the loaded whitelist. Returns a dict with at least a "status" key (SIG_VERIFIED / SIG_UNSIGNED / SIG_INVALID) plus whatever whitelist metadata is available, for display in the UI. Never raises. """ result = {"status": SIG_UNSIGNED, "sha256": None, "signature": None, "signed_at": None} if self._public_key is None or self._whitelist is None: return result entry = self._whitelist.entries.get(name) if entry is None: return result result.update(sha256=entry.sha256, signature=entry.signature, signed_at=entry.signed_at) try: ok = xsign.verify_plugin_on_disk(self._public_key, Path(plugin_path), entry) except Exception: # noqa: BLE001 - never let a verification bug crash attach() logger.error("Signature verification raised for plugin '%s'.", name, exc_info=True) ok = False result["status"] = SIG_VERIFIED if ok else SIG_INVALID return result 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) # Verify the plugin's signature BEFORE we ever execute its code. # Unsigned/tampered plugins still get a listing entry (built # from the manifest alone, which is inert JSON) but their .py # file is never imported, and they can never be run. sig = self._verify_signature(name, plugin_path, module_name) if sig["status"] != SIG_VERIFIED: if sig["status"] == SIG_INVALID: logger.error( "Plugin '%s' failed signature verification (tampered or bad " "signature); attaching as permanently disabled.", name, ) else: logger.warning( "Plugin '%s' has no valid whitelist entry; attaching as " "permanently disabled.", name, ) self.plugins[name] = { "instance": None, "running": False, "info": plugin_info, "autorun": False, "module_key": None, "disabled": True, "signature": sig, "pre_load_thread": None, } return # 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, "disabled": False, "signature": sig, "pre_load_thread": None, } 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) self.plugins[name]["pre_load_thread"] = thread 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 if entry.get("disabled"): logger.error("Plugin '%s' is disabled (failed signature verification) and cannot run.", name) return entry["running"] = True instance = entry["instance"] try: return instance.run(*args, **kwargs) except Exception: # noqa: BLE001 logger.error("Plugin '%s' raised during run.", name, exc_info=True) with self._lock: 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 instance = entry["instance"] try: instance.stop() except Exception: # noqa: BLE001 logger.error("Plugin '%s' raised during stop.", name, exc_info=True) def detach(self, name: str, *, timeout: float = 5.0, force: bool = False) -> None: """Unload a plugin: stop it if running, let it release its own resources via unload(), then drop every reference XtendRSystem holds to it (dict entry, sys.modules entry) so nothing outside the plugin's own cleanup keeps it alive. Protected systems (see __init__) refuse this unless force=True is passed explicitly -- the caller has to opt in on purpose, so a generic "detach" UI action wired up against this system can't accidentally (or maliciously) unload a built-in. """ if self.protected and not force: logger.error( "Refusing to detach '%s': this plugin system is protected " "and cannot be detached from without force=True.", name, ) return with self._lock: entry = self.plugins.pop(name, None) if entry is None: logger.info("Plugin '%s' is not attached.", name) return was_running = entry["running"] entry["running"] = False instance = entry["instance"] thread = entry.get("pre_load_thread") # pre_load() may still be running in its own thread (e.g. detach # called right after attach). Give it a bounded chance to finish # before we call stop()/unload(), so a plugin doesn't get stopped # out from under itself mid pre_load. If it doesn't finish in # time we proceed anyway -- it's a daemon thread and holds its # own reference to the instance, so this is a correctness/race # concern rather than a leak. if thread is not None and thread.is_alive(): thread.join(timeout) if thread.is_alive(): logger.warning( "Plugin '%s' pre_load() did not finish within %.1fs; " "detaching anyway.", name, timeout, ) if instance is not None: if was_running: try: instance.stop() except Exception: # noqa: BLE001 logger.error("Plugin '%s' raised during stop.", name, exc_info=True) try: instance.unload() except Exception: # noqa: BLE001 logger.error("Plugin '%s' raised during unload.", name, exc_info=True) if entry.get("module_key"): sys.modules.pop(entry["module_key"], None) # Drop our own strong references explicitly rather than letting # them idle until this frame unwinds -- entry/instance are the # last references XtendRSystem holds, so this makes the plugin # object (and anything it exclusively owns) collectible the # moment its unload() has actually let go of its own resources. entry["instance"] = None entry["pre_load_thread"] = None logger.info("Detached plugin '%s'.", name) def detach_all(self, *, force: bool = False) -> None: """Detach every currently-attached plugin. Convenience for clean shutdown; equivalent to calling detach() on every plugin name. For a protected system this is a no-op unless force=True (see detach()) -- shutdown code that genuinely needs to tear a protected system down (e.g. the host application closing) passes force=True explicitly; nothing else can. """ if self.protected and not force: logger.error( "Refusing detach_all(): this plugin system is protected " "and cannot be detached from without force=True." ) return with self._lock: names = list(self.plugins.keys()) for name in names: self.detach(name, force=force)