Gated signatures. /JL

xtendr/xtendrsystem.py

@@ -6,9 +6,11 @@ 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.4.0"
+__version__ = "0.5.0"
 
 logger = logging.getLogger("xtendr")
 
@@ -18,6 +20,11 @@ logger = logging.getLogger("xtendr")
 _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.
@@ -30,7 +37,7 @@ class XtendRSystem:
     Example:
     >>> system = XtendRSystem()
     >>> system.version()
-    XtendR v0.4.0
+    XtendR v0.5.0
     >>> system.attach("example_plugin", lambda: None)
     >>> system.run("example_plugin")
     ExamplePlugin is running!
@@ -40,20 +47,81 @@ class XtendRSystem:
     Detached plugin 'example_plugin'.
     """
 
-    def __init__(self, pluginpath="plugins"):
+    def __init__(self, pluginpath="plugins", public_key_path=None, whitelist_path=None, whitelist_passphrase=None):
         self.pluginspath = pluginpath
         self.plugins = {}
         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:
@@ -116,6 +184,33 @@ class XtendRSystem:
 
             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,
+                }
+                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.
@@ -147,6 +242,8 @@ class XtendRSystem:
                 "info": plugin_info,
                 "autorun": False,
                 "module_key": qualified_name,
+                "disabled": False,
+                "signature": sig,
             }
             logger.info("Attached plugin '%s'.", name)
             logger.info("Running pre-load on '%s'.", name)
@@ -167,6 +264,9 @@ class XtendRSystem:
         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
         try:
             return entry["instance"].run(*args, **kwargs)
@@ -194,5 +294,6 @@ class XtendRSystem:
             if entry is None:
                 logger.info("Plugin '%s' is not attached.", name)
                 return
-            sys.modules.pop(entry["module_key"], None)
+            if entry.get("module_key"):
+                sys.modules.pop(entry["module_key"], None)
             logger.info("Detached plugin '%s'.", name)