Added a reload plugin method. /JL

This commit was merged in pull request #5.
This commit is contained in:
2026-09-06 12:39:00 +02:00
parent 818cd44b3d
commit a91efef114
3 changed files with 72 additions and 14 deletions
+1 -1
View File
@@ -1,5 +1,5 @@
# XtendR # XtendR
![Latest Version](https://gitpot.lerk.ing/badges/badge/static?label=Latest+version&message=0.5.2&color=blue) ![Latest Version](https://gitpot.lerk.ing/badges/badge/static?label=Latest+version&message=0.5.3&color=blue)
A very basic Python 3.12 friendly plugin system based on the K.I.S.S principle. A very basic Python 3.12 friendly plugin system based on the K.I.S.S principle.
+1 -1
View File
@@ -6,7 +6,7 @@ if __name__ == "__main__":
setup( setup(
name="XtendR", name="XtendR",
version="0.5.2", version="0.5.3",
packages=find_packages(), packages=find_packages(),
install_requires=[], install_requires=[],
author="Jan Lerking", author="Jan Lerking",
+70 -12
View File
@@ -65,26 +65,23 @@ class XtendRSystem:
self._lock = threading.RLock() self._lock = threading.RLock()
# -- signature verification setup --------------------------------- # -- signature verification setup ---------------------------------
# Paths are kept (not just the loaded key/whitelist objects) so
# reload() can re-read them from disk later -- e.g. after an admin
# re-signs a plugin or adds a whitelist entry for it -- without
# restarting the whole process.
#
# If either the public key or the whitelist can't be loaded, we # If either the public key or the whitelist can't be loaded, we
# fail closed: self._public_key / self._whitelist stay None, and # fail closed: self._public_key / self._whitelist stay None, and
# every plugin will come back as SIG_UNSIGNED (disabled) rather # every plugin will come back as SIG_UNSIGNED (disabled) rather
# than silently skipping verification. This is deliberate -- an # than silently skipping verification. This is deliberate -- an
# admin who wants unsigned plugins to run should not be able to # admin who wants unsigned plugins to run should not be able to
# get there by accident (e.g. a missing/misspelled key path). # get there by accident (e.g. a missing/misspelled key path).
self._public_key_path = public_key_path
self._whitelist_path = whitelist_path
self._whitelist_passphrase = whitelist_passphrase
self._public_key = None self._public_key = None
self._whitelist = None self._whitelist = None
self._reload_verification_material()
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: if self._public_key is None or self._whitelist is None:
logger.warning( logger.warning(
@@ -93,6 +90,30 @@ class XtendRSystem:
pluginpath, pluginpath,
) )
def _reload_verification_material(self) -> None:
"""(Re-)read the public key and whitelist from disk into
self._public_key / self._whitelist. Called once from __init__,
and again from reload() so a freshly-signed plugin or updated
whitelist can be picked up without restarting the process.
Fails closed on error, same as __init__ did: a key/whitelist
that can't be (re)loaded leaves that half of verification as
None rather than keeping around whatever was previously loaded.
"""
self._public_key = None
if self._public_key_path is not None:
try:
self._public_key = xsign.load_public_key(Path(self._public_key_path))
except (OSError, ValueError) as e:
logger.error("Could not load XtendR public key from '%s': %s", self._public_key_path, e)
self._whitelist = None
if self._whitelist_path is not None:
try:
self._whitelist = xsign.Whitelist.load(Path(self._whitelist_path), self._whitelist_passphrase)
except (OSError, ValueError) as e:
logger.error("Could not load XtendR plugin whitelist from '%s': %s", self._whitelist_path, e)
def version(self) -> str: def version(self) -> str:
return "XtendR v" + __version__ return "XtendR v" + __version__
@@ -272,6 +293,43 @@ class XtendRSystem:
self.plugins[name]["pre_load_thread"] = thread self.plugins[name]["pre_load_thread"] = thread
thread.start() thread.start()
def reload(self, name: str, callback=None) -> bool:
"""Re-attempt attaching a plugin that's currently disabled due to
failed signature verification (SIG_UNSIGNED / SIG_INVALID) -- e.g.
after re-signing it or adding/fixing its whitelist entry.
This is specifically a recovery path for that one state, not a
general-purpose re-attach: a plugin that isn't currently attached,
or is attached but not disabled, is left alone and this returns
False without side effects.
The public key and whitelist are re-read from disk before
re-verifying -- __init__ only loads them once, so without this a
freshly-updated whitelist would never actually take effect and
reload() would just repeat the same failed check.
Returns True if the plugin ends up attached and enabled, False if
it's still disabled (or wasn't reloadable to begin with).
"""
with self._lock:
entry = self.plugins.get(name)
if entry is None:
logger.error("Plugin '%s' is not attached; nothing to reload.", name)
return False
if not entry.get("disabled"):
logger.info("Plugin '%s' is not disabled; reload() is a no-op.", name)
return False
# Drop the disabled placeholder so attach() below doesn't just
# see 'already attached' and bail out immediately.
self.plugins.pop(name, None)
self._reload_verification_material()
self.attach(name, callback)
with self._lock:
entry = self.plugins.get(name)
return entry is not None and not entry.get("disabled")
def run(self, name: str, *args, **kwargs): def run(self, name: str, *args, **kwargs):
"""Run the plugin's 'run' method if available.""" """Run the plugin's 'run' method if available."""
with self._lock: with self._lock: