diff --git a/README.md b/README.md index b714631..78188c0 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,5 @@ # 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. diff --git a/setup.py b/setup.py index ff5b722..67c1001 100644 --- a/setup.py +++ b/setup.py @@ -6,7 +6,7 @@ if __name__ == "__main__": setup( name="XtendR", - version="0.5.2", + version="0.5.3", packages=find_packages(), install_requires=[], author="Jan Lerking", diff --git a/xtendr/xtendrsystem.py b/xtendr/xtendrsystem.py index c9d4fd5..4ec0ad8 100644 --- a/xtendr/xtendrsystem.py +++ b/xtendr/xtendrsystem.py @@ -65,26 +65,23 @@ class XtendRSystem: self._lock = threading.RLock() # -- 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 # 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_path = public_key_path + self._whitelist_path = whitelist_path + self._whitelist_passphrase = whitelist_passphrase 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) + self._reload_verification_material() if self._public_key is None or self._whitelist is None: logger.warning( @@ -93,6 +90,30 @@ class XtendRSystem: 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: return "XtendR v" + __version__ @@ -272,6 +293,43 @@ class XtendRSystem: self.plugins[name]["pre_load_thread"] = thread 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): """Run the plugin's 'run' method if available.""" with self._lock: