Lerking/XtendR
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 6 Wiki Activity

Updated. /JL

Browse Source
This commit is contained in:
Lerking
2026-07-27 20:54:00 +02:00
parent ad67becd91
commit c6b91303ad
5 changed files with 152 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
xtendr/xtendrsystem.py
+116 -22
View File
@@ -10,7 +10,7 @@ from pathlib import Path
from xtendr.xtendrbase import XtendRBase
from xtendr import signing as xsign
__version__ = "0.5.0"
__version__ = "0.5.1"
logger = logging.getLogger("xtendr")
@@ -37,7 +37,7 @@ class XtendRSystem:
Example:
>>> system = XtendRSystem()
>>> system.version()
XtendR v0.5.0
XtendR v0.5.1
>>> system.attach("example_plugin", lambda: None)
>>> system.run("example_plugin")
ExamplePlugin is running!
@@ -45,11 +45,23 @@ class XtendRSystem:
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):
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 ---------------------------------
@@ -208,6 +220,7 @@ class XtendRSystem:
"module_key": None,
"disabled": True,
"signature": sig,
"pre_load_thread": None,
}
return
@@ -244,6 +257,7 @@ class XtendRSystem:
"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)
@@ -255,45 +269,125 @@ class XtendRSystem:
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
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 entry["instance"].run(*args, **kwargs)
return instance.run(*args, **kwargs)
except Exception: # noqa: BLE001
logger.error("Plugin '%s' raised during run.", name, exc_info=True)
entry["running"] = False
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
if entry is None or not entry["running"]:
logger.info("Plugin '%s' is not running.", name)
return
entry["running"] = False
instance = entry["instance"]
try:
entry["instance"].stop()
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."""
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
if entry.get("module_key"):
sys.modules.pop(entry["module_key"], None)
logger.info("Detached plugin '%s'.", name)
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)