Merge pull request #10874 from spesmilo/plugin_zipfile

### electrum/gui/qt/plugins_dialog.py
@@ -1,7 +1,5 @@
 from typing import TYPE_CHECKING, Optional
 from functools import partial
-import shutil
-import os
 
 from PyQt6.QtWidgets import QLabel, QVBoxLayout, QHBoxLayout, QGridLayout, QPushButton, QWidget, QScrollArea, \
     QFormLayout, QFileDialog, QMenu, QApplication, QMessageBox
@@ -25,7 +23,9 @@
 
 class PluginDialog(WindowModalDialog):
 
-    def __init__(self, name, metadata, status_button: Optional['PluginStatusButton'], window: 'PluginsDialog'):
+    def __init__(self, name, metadata, status_button: Optional['PluginStatusButton'], window: 'PluginsDialog',
+                 *, is_upgrade: bool = False):
+        """If is_upgrade, `metadata` describes a new version of an installed plugin."""
         display_name = metadata.get('fullname', '')
         author = metadata.get('author', '')
         description = metadata.get('description', '')
@@ -46,8 +46,7 @@ def __init__(self, name, metadata, status_button: Optional['PluginStatusButton']
         name_label = IconLabel(text=display_name, reverse=True)
         if icon_path:
             name_label.icon_size = 64
-            icon = read_QIcon_from_bytes(self.plugins.read_file(name, icon_path))
-            name_label.setIcon(icon)
+            self.window.maybe_set_icon(name_label, name, icon_path, manifest=metadata if is_upgrade else None)
         vbox.addWidget(name_label)
         vbox.addStretch()
         vbox.addWidget(WWLabel(description))  # must be plain text: don't parse untrusted text as rich-text
@@ -57,6 +56,8 @@ def __init__(self, name, metadata, status_button: Optional['PluginStatusButton']
             form.addRow(QLabel(_('Author') + ':'), QLabel(author))
         if version:
             form.addRow(QLabel(_('Version') + ':'), QLabel(version))
+        if is_upgrade and (installed_version := self.plugins.get_metadata(name).get('version')):
+            form.addRow(QLabel(_('Installed version') + ':'), QLabel(installed_version))
         if zip_hash:
             form.addRow(QLabel('Hash [sha256]:'), WWLabel(insert_spaces(zip_hash, 8)))
         if requires:
@@ -70,10 +71,15 @@ def __init__(self, name, metadata, status_button: Optional['PluginStatusButton']
         p = self.plugins.get(name)
         is_enabled = p and p.is_enabled()
         is_external = self.plugins.is_external(name)
-        if is_external:
+        if is_upgrade:
+            if zip_hash != self.plugins.get_metadata(name)['zip_hash_sha256']:
+                upgrade_button = QPushButton(_('Upgrade'))
+                upgrade_button.clicked.connect(self.do_upgrade)
+                buttons.insert(0, upgrade_button)
+        elif is_external:
             is_authorized = self.plugins.is_authorized(name)
             if status_button is not None:
-                # status_button is None when called from add_external_plugin
+                # status_button is None when called from add_plugin_dialog
                 remove_button = QPushButton('')
                 remove_button.clicked.connect(self.do_remove)
                 remove_button.setText(_('Remove'))
@@ -88,7 +94,7 @@ def __init__(self, name, metadata, status_button: Optional['PluginStatusButton']
             toggle_button.clicked.connect(self.do_toggle)
             buttons.insert(0, toggle_button)
         # add settings button
-        if p and p.requires_settings() and p.is_enabled():
+        if p and p.requires_settings() and p.is_enabled() and not is_upgrade:
             settings_button = EnterButton(
                 _('Settings'),
                 partial(p.settings_dialog, self))
@@ -117,8 +123,11 @@ def do_authorize(self):
         privkey = self.window.get_plugins_privkey()
         if not privkey:
             return
-        filename = self.plugins.zip_plugin_path(self.name)
-        self.window.plugins.authorize_plugin(self.name, filename, privkey)
+        try:
+            self.window.plugins.authorize_plugin(self.name, privkey)
+        except Exception as e:
+            self.show_error(f"{e}")
+            return
         self.window.plugins.enable(self.name)
         d = self.plugins.get_metadata(self.name)
         if details := d.get('registers_keystore'):
@@ -127,6 +136,17 @@ def do_authorize(self):
             self.status_button.update()
         self.accept()
 
+    def do_upgrade(self):
+        privkey = self.window.get_plugins_privkey()
+        if not privkey:
+            return
+        try:
+            self.plugins.upgrade_external_plugin(self.metadata, privkey)
+        except Exception as e:
+            self.show_error(f"{e}")
+            return
+        self.accept()
+
 
 class PluginStatusButton(QPushButton):
 
@@ -285,47 +305,47 @@ def add_plugin_dialog(self):
         filename, __ = QFileDialog.getOpenFileName(self, _("Select your plugin zipfile"), "", "*.zip")
         if not filename:
             return
-        plugins_dir = self.plugins.get_external_plugin_dir()
-        path = os.path.join(plugins_dir, os.path.basename(filename))
-        if os.path.exists(path):
-            self.show_warning(_('Plugin already installed.'))
-            return
         try:
-            shutil.copyfile(filename, path)
-        except OSError as e:
-            self.show_error(_("Could not copy plugin file {} into directory {}:\n\n{}").format(
-                filename,
-                path,
-                str(e)
-            ))
-            return
-        self._try_add_external_plugin_from_path(path)
-
-    def _try_add_external_plugin_from_path(self, path: str):
-        try:
-            success = self.add_external_plugin(path)
+            manifest = self.plugins.read_manifest(filename)
+            name = manifest['name']
         except Exception as e:
             self._logger.exception("")
             self.show_error(f"{e}")
-            success = False
-        if not success:
-            try:
-                os.unlink(path)
-            except FileNotFoundError:
-                self._logger.debug("", exc_info=True)
-
-    def add_external_plugin(self, path):
-        manifest = self.plugins.read_manifest(path)
-        name = manifest['name']
-        self.plugins.external_plugin_metadata[name] = manifest
+            return
+        if self.plugins.is_external(name):
+            self.upgrade_plugin(manifest)
+            return
+        if self.plugins.is_installed(name):
+            self.show_warning(_("Plugin {} already installed.").format(name))
+            return
+        # the file is copied into the plugins directory if the user clicks 'Install'
+        self.plugins.add_external_plugin_metadata(manifest)
         d = PluginDialog(name, manifest, None, self)
         if not d.exec():
-            self.plugins.external_plugin_metadata.pop(name)
-            return False
+            self.plugins.remove_external_plugin_metadata(name)
+            return
         if self.gui_object:
             self.gui_object.reload_windows()
         self.show_list()
-        return True
+
+    def upgrade_plugin(self, manifest: dict):
+        d = PluginDialog(manifest['name'], manifest, None, self, is_upgrade=True)
+        if not d.exec():
+            return
+        self.show_message(_('Please restart Electrum to use the new version.'))
+        self.show_list()
+
+    def maybe_set_icon(self, label, name, icon_path, *, manifest: dict = None):
+        """Sets the icon of the installed plugin, or of the zip described by `manifest`."""
+        try:
+            if manifest:
+                icon_bytes = self.plugins.read_zip_file(manifest, icon_path)
+            else:
+                icon_bytes = self.plugins.read_file(name, icon_path)
+            icon = read_QIcon_from_bytes(icon_bytes)
+        except Exception:
+            icon = read_QIcon('warning.png')
+        label.setIcon(icon)
 
     def show_list(self):
         descriptions = self.plugins.descriptions
@@ -346,8 +366,7 @@ def show_list(self):
             label = IconLabel(text=display_name, reverse=True)
             icon_path = metadata.get('icon')
             if icon_path:
-                icon = read_QIcon_from_bytes(self.plugins.read_file(name, icon_path))
-                label.setIcon(icon)
+                self.maybe_set_icon(label, name, icon_path)
             label.status_button = PluginStatusButton(self, name)
             grid.addWidget(label, i, 0)
             grid.addWidget(label.status_button, i, 1)
@@ -374,6 +393,7 @@ def uninstall_plugin(self, name):
         if self.gui_object:
             self.gui_object.reload_windows()
         self.show_list()
+        self.show_message(_('Please restart Electrum to finish removing the plugin.'))
         self.bring_to_front()
 
     def bring_to_front(self):

### electrum/plugin.py
@@ -23,6 +23,7 @@
 # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
 
+import io
 import json
 import os
 import pkgutil
@@ -37,7 +38,6 @@
                     Dict, Iterable, List, Sequence, Callable, TypeVar, Mapping)
 import concurrent
 from concurrent.futures import Future
-import zipimport
 from functools import wraps, partial
 from itertools import chain
 
@@ -55,6 +55,7 @@
 from .logging import get_logger, Logger
 from .crypto import sha256
 from .network import Network
+from .zip_importer import MemoryZipImporter
 
 if TYPE_CHECKING:
     from .hw_wallet import HW_PluginBase, HardwareClientBase, HardwareHandlerBase
@@ -70,6 +71,14 @@
 
 PLUGIN_PASSWORD_VERSION = 1
 
+class IncorrectPluginHash(Exception):
+    pass
+
+
+# cached importers
+_zip_importers = {}  # type: Dict[str, MemoryZipImporter]  # see _get_zip_importer()
+_zip_importers_lock = threading.RLock()
+
 
 class Plugins(DaemonThread):
 
@@ -506,9 +515,21 @@ async def download_external_plugin(self, url: str) -> str:
                             fd.write(chunk)
         return path
 
+    def _read_check_bytes(self, path: str, *, expected_hash: Optional[bytes] = None) -> bytes:
+        # Read the file and compare the hash to expectation
+        MAX_SIZE = 10_000_000
+        with open(path, 'rb') as f:
+            blob = f.read(MAX_SIZE + 1)
+        if len(blob) > MAX_SIZE:
+            raise Exception(f'Plugin file too large {path}')
+        if expected_hash is not None and sha256(blob) != expected_hash:
+            raise IncorrectPluginHash(f'Plugin file {path} has been modified (sha256 mismatch)')
+        return blob
+
     def read_manifest(self, path) -> dict:
-        """ return json dict """
-        with zipfile_lib.ZipFile(path) as file:
+        """ return plugin manifest """
+        blob = self._read_check_bytes(path, expected_hash=None)
+        with zipfile_lib.ZipFile(io.BytesIO(blob)) as file:
             for filename in file.namelist():
                 if filename.endswith('manifest.json'):
                     break
@@ -519,9 +540,19 @@ def read_manifest(self, path) -> dict:
                 manifest['path'] = path  # external, path of the zipfile
                 manifest['dirname'] = os.path.dirname(filename)  # internal
                 manifest['is_zip'] = True
-                manifest['zip_hash_sha256'] = get_file_hash256(path).hex()
+                manifest['zip_hash_sha256'] = sha256(blob).hex()
                 return manifest
 
+    @staticmethod
+    def _is_version_compatible(d: dict) -> bool:
+        min_version = d.get('min_electrum_version')
+        if min_version and StrictVersion(min_version) > StrictVersion(ELECTRUM_VERSION):
+            return False
+        max_version = d.get('max_electrum_version')
+        if max_version and StrictVersion(max_version) < StrictVersion(ELECTRUM_VERSION):
+            return False
+        return True
+
     def zip_plugin_path(self, name) -> str:
         path = self.get_metadata(name)['path']
         filename = os.path.basename(path)
@@ -550,13 +581,8 @@ def find_zip_plugins(self, pkg_path: str, external: bool):
                 continue
             if self.cmd_only and not self.config.get(f'plugins.{name}.enabled'):
                 continue
-            min_version = d.get('min_electrum_version')
-            if min_version and StrictVersion(min_version) > StrictVersion(ELECTRUM_VERSION):
-                self.logger.info(f"version mismatch for zip plugin {filename}", exc_info=True)
-                continue
-            max_version = d.get('max_electrum_version')
-            if max_version and StrictVersion(max_version) < StrictVersion(ELECTRUM_VERSION):
-                self.logger.info(f"version mismatch for zip plugin {filename}", exc_info=True)
+            if not self._is_version_compatible(d):
+                self.logger.info(f"version mismatch for zip plugin {filename}")
                 continue
 
             if not self.cmd_only:
@@ -588,11 +614,14 @@ def load_plugin(self, name) -> 'BasePlugin':
         else:
             raise Exception(f"could not find plugin {name!r}")
 
+    def base_module_name(self, name: str) -> str:
+        return ('electrum_external_plugins.' if self.is_external(name) else 'electrum.plugins.') + name
+
     def maybe_load_plugin_init_method(self, name: str) -> None:
         """Loads the __init__.py module of the plugin if it is not already loaded."""
         if not self.is_authorized(name):
             return
-        base_name = ('electrum_external_plugins.' if self.is_external(name) else 'electrum.plugins.') + name
+        base_name = self.base_module_name(name)
         if base_name not in sys.modules:
             metadata = self.get_metadata(name)
             is_zip = metadata.get('is_zip', False)
@@ -605,9 +634,13 @@ def maybe_load_plugin_init_method(self, name: str) -> None:
                 else:
                     init_spec = importlib.util.find_spec(base_name)
             else:
-                zipfile = zipimport.zipimporter(metadata['path'])
-                dirname = metadata['dirname']
-                init_spec = zipfile.find_spec(dirname)
+                # This registers the importer on sys.meta_path
+                importer = self._get_zip_importer(name)
+                if importer is None:
+                    raise Exception(f"plugin {name!r} is not authorized")
+                init_spec = importer.find_spec(base_name)
+                if init_spec is None:
+                    raise Exception(f"no __init__.py for plugin {name!r} in {metadata['path']}")
 
             self.exec_module_from_spec(init_spec, base_name)
 
@@ -618,12 +651,7 @@ def load_plugin_by_name(self, name: str) -> Optional['BasePlugin']:
             return self.plugins[name]
         # if the plugin was not enabled on startup the init module hasn't been loaded yet
         self.maybe_load_plugin_init_method(name)
-        is_external = self.is_external(name)
-        if not is_external:
-            full_name = f'electrum.plugins.{name}.{self.gui_name}'
-        else:
-            full_name = f'electrum_external_plugins.{name}.{self.gui_name}'
-
+        full_name = f'{self.base_module_name(name)}.{self.gui_name}'
         spec = importlib.util.find_spec(full_name)
         if spec is None:
             raise RuntimeError(f"{self.gui_name} implementation for {name} plugin not found")
@@ -646,7 +674,18 @@ def derive_privkey(pw: str, salt: bytes) -> ECPrivkey:
         secret = pbkdf2_hmac('sha256', pw.encode('utf-8'), salt, iterations=10**5)
         return ECPrivkey(secret)
 
+    def add_external_plugin_metadata(self, manifest: dict) -> None:
+        """Registers the metadata of a newly installed external plugin."""
+        name = manifest['name']
+        assert name not in self.external_plugin_metadata
+        self.external_plugin_metadata[name] = manifest
+
+    def remove_external_plugin_metadata(self, name: str) -> None:
+        """Unregisters an external plugin that did not end up being installed."""
+        self.external_plugin_metadata.pop(name, None)
+
     def uninstall(self, name: str):
+        self.disable(name)
         if self.config.get(f'plugins.{name}'):
             self.config.set_key(f'plugins.{name}', None)
         if name in self.external_plugin_metadata:
@@ -668,6 +707,33 @@ def is_installed(self, name) -> bool:
         """an external plugin may be installed but not authorized """
         return (name in self.internal_plugin_metadata or name in self.external_plugin_metadata)
 
+    def _get_zip_importer(self, name: str, *, cache_only: bool = False) -> Optional[MemoryZipImporter]:
+        """Returns an in-memory copy of a zip plugin's archive, or None if the
+        plugin is not (or no longer) authorized.
+        """
+        metadata = self.get_metadata(name)
+        if metadata is None or not metadata.get('is_zip', False):
+            return None
+        is_external = self.is_external(name)
+        # internal plugins ship inside the application bundle, and are not signed separately
+        if is_external and not self.is_authorized(name):
+            return None
+        with _zip_importers_lock:
+            if (cached := _zip_importers.get(name)) is not None:
+                return cached
+            elif cache_only:
+                return None
+            blob = self._read_check_bytes(metadata['path'], expected_hash=bytes.fromhex(metadata['zip_hash_sha256']))
+            archive_path = self.zip_plugin_path(name)
+            importer = MemoryZipImporter(
+                blob,
+                root_name=self.base_module_name(name),
+                prefix=metadata['dirname'],
+                archive_path=archive_path,
+            )
+            _zip_importers[name] = importer.install()
+            return importer
+
     def is_authorized(self, name) -> bool:
         if name in self.internal_plugin_metadata:
             return True
@@ -676,25 +742,76 @@ def is_authorized(self, name) -> bool:
         pubkey_bytes, salt = self.get_pubkey_bytes()
         if not pubkey_bytes:
             return False
-        if not self.is_plugin_zip(name):
+        metadata = self.external_plugin_metadata[name]
+        if not metadata.get('is_zip'):
             return False
-        filename = self.zip_plugin_path(name)
-        plugin_hash = get_file_hash256(filename)
+        if metadata['path'] != self.zip_plugin_path(name):
+            # not installed yet, or was manually deleted
+            return False
+        hex_hash = metadata['zip_hash_sha256']
         sig = self.config.get(f'plugins.{name}.authorized')
         if not sig:
             return False
-        pubkey = ECPubkey(pubkey_bytes)
-        return pubkey.ecdsa_verify(bytes.fromhex(sig), plugin_hash)
+        try:
+            verified = ECPubkey(pubkey_bytes).ecdsa_verify(bytes.fromhex(sig), bytes.fromhex(hex_hash))
+        except Exception:
+            self.logger.info(f"malformed signature for plugin {name!r}", exc_info=True)
+            verified = False
+        return verified
 
-    def authorize_plugin(self, name: str, filename, privkey: ECPrivkey):
+    def _sign_plugin_hash(self, name: str, privkey: ECPrivkey) -> None:
         pubkey_bytes, salt = self.get_pubkey_bytes()
         assert pubkey_bytes == privkey.get_public_key_bytes()
-        plugin_hash = get_file_hash256(filename)
+        plugin_hash = bytes.fromhex(self.get_metadata(name)['zip_hash_sha256'])
         sig = privkey.ecdsa_sign(plugin_hash)
-        value = sig.hex()
-        self.config.set_key(f'plugins.{name}.authorized', value)
+        self.config.set_key(f'plugins.{name}.authorized', sig.hex())
+
+    def authorize_plugin(self, name: str, privkey: ECPrivkey):
+        metadata = self.get_metadata(name)
+        if metadata['path'] != self.zip_plugin_path(name):
+            # new plugin, not yet in the plugins directory
+            metadata['path'] = self.copy_external_plugin_file(metadata)
+        self._sign_plugin_hash(name, privkey)
         self.config.set_key(f'plugins.{name}.enabled', True)
 
+    def copy_external_plugin_file(self, manifest: dict, *, replace: Optional[str] = None) -> str:
+        """Copies the zip described by `manifest` (as returned by read_manifest) into
+        the external plugins directory, keeping its name, and returns its new path.
+        If `replace` is given, that file is removed first.
+        """
+        # the bytes we write are the bytes described by the manifest
+        blob = self._read_check_bytes(manifest['path'], expected_hash=bytes.fromhex(manifest['zip_hash_sha256']))
+        path = os.path.join(self.get_external_plugin_dir(), os.path.basename(manifest['path']))
+        if path != replace and os.path.exists(path):
+            raise FileExistsError(f"Plugin file {path} already exists")
+        if replace:
+            # Remove old file first
+            os.unlink(replace)
+        with open(path, 'wb') as f:
+            f.write(blob)
+        return path
+
+    def upgrade_external_plugin(self, manifest: dict, privkey: ECPrivkey) -> None:
+        """Replaces the zip of an installed external plugin with the one described
+        by `manifest` (as returned by read_manifest), and signs it. The new file
+        keeps its own name, which often contains the version, and the old file
+        is removed.
+
+        This does not go through uninstall(): the 'plugins.<name>' config subtree is
+        kept, and so is the plugin's wallet data (see WalletDB.prune_uninstalled_plugin_data).
+        The client must be restarted: we cannot undo the side effects of the old
+        code, which is imported on startup if the plugin is enabled. Until then,
+        the old code keeps running from memory.
+        """
+        name = manifest['name']
+        assert self.is_external(name) and self.is_plugin_zip(name), name
+        if not self._is_version_compatible(manifest):
+            raise Exception(f"plugin {name!r} is not compatible with Electrum {ELECTRUM_VERSION}")
+        # If we stop before signing, the plugin is missing or unauthorized, but its config is kept.
+        path = self.copy_external_plugin_file(manifest, replace=self.zip_plugin_path(name))
+        self.external_plugin_metadata[name] = dict(manifest, path=path)
+        self._sign_plugin_hash(name, privkey)
+
     def enable(self, name: str) -> 'BasePlugin':
         self.config.enable_plugin(name)
         p = self.get(name)
@@ -800,24 +917,26 @@ def run(self):
 
     def read_file(self, name: str, filename: str) -> bytes:
         if self.is_plugin_zip(name):
-            plugin_filename = self.zip_plugin_path(name)
-            metadata = self.external_plugin_metadata[name]
-            dirname = metadata['dirname']
-            with zipfile_lib.ZipFile(plugin_filename) as myzip:
-                with myzip.open("/".join([dirname,filename])) as myfile:
-                    return myfile.read()
+            if importer := self._get_zip_importer(name, cache_only=True):
+                return importer.read(filename)
+            # Plugins not in authorized list
+            # We allow reading files without importing code (eg to display icon)
+            return self.read_zip_file(self.get_metadata(name), filename)
         elif name in self.internal_plugin_metadata:
             path = os.path.join(os.path.dirname(__file__), 'plugins', name, filename)
             with open(path, 'rb') as myfile:
                 return myfile.read()
         else:
             raise Exception(f"plugin not found: {name!r}")
 
-
-def get_file_hash256(path: str) -> bytes:
-    """Get the sha256 hash of a file, similar to `sha256sum`."""
-    with open(path, 'rb') as f:
-        return sha256(f.read())
+    def read_zip_file(self, manifest: dict, filename: str) -> bytes:
+        """Reads a file from the zip described by `manifest`, without importing its code."""
+        dirname = manifest['dirname']
+        blob = self._read_check_bytes(manifest['path'], expected_hash=bytes.fromhex(manifest['zip_hash_sha256']))
+        member = "/".join([dirname, filename]) if dirname else filename
+        with zipfile_lib.ZipFile(io.BytesIO(blob)) as myzip:
+            with myzip.open(member) as myfile:
+                return myfile.read()
 
 
 def hook(func):

### electrum/zip_importer.py
@@ -0,0 +1,187 @@
+#!/usr/bin/env python
+#
+# Electrum - lightweight Bitcoin client
+# Copyright (C) 2026 The Electrum developers
+#
+# Permission is hereby granted, free of charge, to any person
+# obtaining a copy of this software and associated documentation files
+# (the "Software"), to deal in the Software without restriction,
+# including without limitation the rights to use, copy, modify, merge,
+# publish, distribute, sublicense, and/or sell copies of the Software,
+# and to permit persons to whom the Software is furnished to do so,
+# subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be
+# included in all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
+# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
+# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+"""Importer for a zip archive held in memory.
+
+Written for the external plugin loader, which authorizes an archive by
+verifying an ECDSA signature over its sha256. If the archive is then read from
+disk *again* -- to import a module, or to read an icon -- someone who can write
+to that directory can replace the file between the check and the read, and the
+code that gets executed is not the code that was authorized. `zipimport` makes
+this especially easy to hit, because it re-opens the archive on every single
+module load, and it never verifies the CRC of what it reads.
+
+`MemoryZipImporter` closes that window. The archive is read into memory once,
+and every module and resource is served from those bytes. `sha256` is the
+digest of the very bytes being held, so a caller that verifies it has no second
+read left to race.
+
+This is not a general replacement for `zipimport`: it mounts exactly one
+package, deliberately never writes bytecode, and does not implement
+`ResourceReader` or nested archives.
+"""
+
+import io
+import importlib.abc
+import importlib.util
+import sys
+import threading
+import zipfile as zipfile_lib
+from typing import Dict, Optional, Tuple
+import hashlib
+
+
+
+class MemoryZipImporter(importlib.abc.SourceLoader, importlib.abc.MetaPathFinder):
+    """A meta path finder and source loader backed by an in-memory zip archive.
+
+    `prefix` is the directory inside the archive that holds the package (i.e.
+    the directory containing its `__init__.py`), and it is mounted under the
+    module name `root_name`. `archive_path` is used for display only: it
+    ends up in `__file__` and in tracebacks, and is never opened.
+    """
+
+    def __init__(self, blob: bytes, *, root_name: str, prefix: str = '', archive_path: str = '<memory>'):
+        self._sha256 = bytes(hashlib.sha256(blob).digest())
+        self._zip = zipfile_lib.ZipFile(io.BytesIO(blob))
+        self._root_name = root_name
+        self._prefix = (prefix.strip('/') + '/') if prefix.strip('/') else ''
+        self._archive_path = archive_path
+        # ZipFile is not thread-safe, and modules can be imported from any thread
+        self._lock = threading.RLock()
+        self._modules = {}  # type: Dict[str, Tuple[str, bool]]  # module name -> (member, is_package)
+        self._filenames = {}  # type: Dict[str, str]  # synthetic __file__ -> member
+        for member in self._zip.namelist():
+            entry = self._module_for_member(member)
+            if entry is None:
+                continue
+            fullname, is_pkg = entry
+            self._modules[fullname] = (member, is_pkg)
+            self._filenames[self._synthetic_path(member)] = member
+
+    def _module_for_member(self, member: str) -> Optional[Tuple[str, bool]]:
+        if not member.startswith(self._prefix) or not member.endswith('.py'):
+            return None
+        rel = member[len(self._prefix):]
+        if rel == '__init__.py':
+            return self._root_name, True
+        elif rel.endswith('/__init__.py'):
+            subname, is_pkg = rel[:-len('/__init__.py')], True
+        else:
+            subname, is_pkg = rel[:-len('.py')], False
+        parts = subname.split('/')
+        # rejects '', '..', and anything else that is not a legal module name
+        if not all(part.isidentifier() for part in parts):
+            return None
+        return self._root_name + '.' + '.'.join(parts), is_pkg
+
+    def _synthetic_path(self, member: str) -> str:
+        return self._archive_path + '/' + member
+
+    def read(self, filename: str) -> bytes:
+        """Reads a file from the archive, relative to `prefix`."""
+        return self._read_member(self._prefix + filename.lstrip('/'))
+
+    def _read_member(self, member: str) -> bytes:
+        with self._lock:
+            try:
+                # note: ZipFile.read() verifies the member CRC, zipimport does not
+                return self._zip.read(member)
+            except KeyError:
+                raise FileNotFoundError(f"{member!r} not found in {self._archive_path}") from None
+
+    # --- MetaPathFinder ---
+
+    def find_spec(self, fullname, path=None, target=None):
+        entry = self._modules.get(fullname)
+        if entry is None:
+            return None
+        member, is_pkg = entry
+        spec = importlib.util.spec_from_loader(fullname, self, is_package=is_pkg)
+        if is_pkg:
+            # Empty on purpose. This finder answers for every
+            # submodule, so __path__ is never consulted while we are
+            # installed. If this method is bypassed we want submodule
+            # lookup to fail with ModuleNotFoundError, rather than
+            # have PathFinder silently resolve a path on disk.
+            spec.submodule_search_locations = []
+        return spec
+
+    # --- SourceLoader ---
+
+    def is_package(self, fullname):
+        try:
+            return self._modules[fullname][1]
+        except KeyError:
+            raise ImportError(f"{fullname!r} is not in {self._archive_path}", name=fullname) from None
+
+    def get_filename(self, fullname):
+        try:
+            return self._synthetic_path(self._modules[fullname][0])
+        except KeyError:
+            raise ImportError(f"{fullname!r} is not in {self._archive_path}", name=fullname) from None
+
+    def get_data(self, path):
+        member = self._filenames.get(path)
+        if member is None:
+            # a resource, addressed by synthetic path or archive-relative name
+            if path.startswith(self._archive_path + '/'):
+                path = path[len(self._archive_path) + 1:]
+                if not path.startswith(self._prefix):
+                    raise FileNotFoundError(path)
+                path = path[len(self._prefix):]
+            member = self._prefix + path.lstrip('/')
+        return self._read_member(member)
+
+    # note: we do not override SourceLoader.path_stats(), whose default raises
+    # OSError. That disables bytecode caching, so we never write .pyc files
+    # (there is no directory to write them to anyway).
+
+    # --- lifecycle ---
+
+    def install(self) -> 'MemoryZipImporter':
+        """Registers this finder in sys.meta_path.
+
+        Prepended, so that it is consulted before PathFinder: the package
+        `__path__` we hand out must never be resolved through sys.path_hooks,
+        which would read the archive from disk again. This cannot shadow
+        anything else, as find_spec() only answers for `root_name` and its
+        submodules.
+        """
+        if self not in sys.meta_path:
+            sys.meta_path.insert(0, self)
+        return self
+
+    def uninstall(self) -> None:
+        if self in sys.meta_path:
+            sys.meta_path.remove(self)
+
+    def close(self) -> None:
+        self.uninstall()
+        with self._lock:
+            self._zip.close()
+
+    def __repr__(self):
+        return f"<MemoryZipImporter {self._root_name} at {self._archive_path} modules={len(self._modules)}>"

### tests/test_plugin.py
@@ -0,0 +1,154 @@
+import os
+import sys
+
+from unittest import mock
+
+from electrum import util
+from electrum import plugin as plugin_module
+from electrum.crypto import sha256
+from electrum.plugin import Plugins, IncorrectPluginHash
+from electrum.zip_importer import MemoryZipImporter
+from electrum.simple_config import SimpleConfig
+
+from electrum_ecc import ECPrivkey
+
+from . import ElectrumTestCase
+from .test_zip_importer import PLUGIN_NAME, make_plugin_zip
+
+
+class PluginLoaderTestCase(ElectrumTestCase):
+    """Tests that an authorized plugin is loaded from the bytes whose signature
+    was verified, and not from whatever happens to be on disk at import time."""
+
+    def setUp(self):
+        super().setUp()
+        self.config = SimpleConfig({'electrum_path': self.electrum_path})
+        self.privkey = ECPrivkey(sha256(b'plugin zip unit test key'))
+        self.plugins_dir = os.path.join(self.electrum_path, 'plugins')
+        util.make_dir(self.plugins_dir)
+        self.zip_path = os.path.join(self.plugins_dir, f'{PLUGIN_NAME}.zip')
+        self._patcher = mock.patch.object(
+            Plugins, 'get_pubkey_bytes',
+            lambda _self: (self.privkey.get_public_key_bytes(), bytes(32)))
+        self._patcher.start()
+        self.plugins = None
+
+    def tearDown(self):
+        self._patcher.stop()
+        self._stop_plugins()
+        super().tearDown()
+
+    def _stop_plugins(self) -> None:
+        if self.plugins is not None:
+            self.plugins.stop()
+            self.plugins.stopped_event.wait()
+            self.plugins = None
+        # the import system is global state; undo what the test did to it
+        for modname in [m for m in sys.modules if m.startswith('electrum_external_plugins')]:
+            del sys.modules[modname]
+        for finder in [f for f in sys.meta_path if isinstance(f, MemoryZipImporter)]:
+            sys.meta_path.remove(finder)
+        plugin_module._zip_importers.clear()
+
+    def _start_plugins(self) -> Plugins:
+        self.plugins = Plugins(self.config, gui_name='cmdline')
+        return self.plugins
+
+    def _authorize(self, plugins: Plugins) -> None:
+        plugins.authorize_plugin(PLUGIN_NAME, self.privkey)
+
+    def _replace_zip_on_disk(self, secret: str = 'evil') -> None:
+        evil = os.path.join(self.electrum_path, 'evil.zip')
+        make_plugin_zip(evil, secret=secret)
+        os.replace(evil, self.zip_path)
+
+    # --- the regression this guards against ---
+
+    def test_import_raises_after_file_replaced(self):
+        make_plugin_zip(self.zip_path, secret='benign')
+        plugins = self._start_plugins()
+        self._authorize(plugins)
+        self.assertTrue(plugins.is_authorized(PLUGIN_NAME))
+        self._replace_zip_on_disk('evil')
+        with self.assertRaises(IncorrectPluginHash):
+            plugin = plugins.load_plugin_by_name(PLUGIN_NAME)
+
+    def test_read_file_raises_after_file_replaced(self):
+        make_plugin_zip(self.zip_path, secret='benign')
+        plugins = self._start_plugins()
+        self._authorize(plugins)
+        self._replace_zip_on_disk('evil')
+        with self.assertRaises(IncorrectPluginHash):
+            plugins.read_file(PLUGIN_NAME, 'icon.txt')
+
+    def test_tampered_plugin_is_not_authorized(self):
+        make_plugin_zip(self.zip_path, secret='benign')
+        plugins = self._start_plugins()
+        self._authorize(plugins)
+        # replace on disk
+        self._replace_zip_on_disk('evil')
+        self.assertTrue(plugins.is_authorized(PLUGIN_NAME))
+        # as if electrum had been restarted
+        plugins.stop()
+        plugins = self._start_plugins()
+        self.assertFalse(plugins.is_authorized(PLUGIN_NAME))
+        self.assertIsNone(plugins.load_plugin_by_name(PLUGIN_NAME))
+
+    def test_unsigned_plugin_is_not_authorized(self):
+        make_plugin_zip(self.zip_path, secret='benign')
+        plugins = self._start_plugins()
+        self.assertFalse(plugins.is_authorized(PLUGIN_NAME))
+        self.assertIsNone(plugins.load_plugin_by_name(PLUGIN_NAME))
+
+    def test_manifest_hash_describes_the_bytes_it_was_read_from(self):
+        blob = make_plugin_zip(self.zip_path, secret='benign')
+        plugins = self._start_plugins()
+        manifest = plugins.read_manifest(self.zip_path)
+        self.assertEqual(sha256(blob).hex(), manifest['zip_hash_sha256'])
+
+    # --- upgrade ---
+
+    def _download_new_version(self, secret: str) -> dict:
+        path = os.path.join(self.electrum_path, f'{PLUGIN_NAME}-{secret}.zip')
+        make_plugin_zip(path, secret=secret)
+        return self.plugins.read_manifest(path)
+
+    def test_upgrade_keeps_config(self):
+        make_plugin_zip(self.zip_path, secret='v1')
+        plugins = self._start_plugins()
+        self._authorize(plugins)
+        plugins.disable(PLUGIN_NAME)
+        plugins.upgrade_external_plugin(self._download_new_version('v2'), self.privkey)
+        # what WalletDB.prune_uninstalled_plugin_data() looks at
+        self.assertIn(PLUGIN_NAME, self.config.get_installed_plugins())
+        self.assertFalse(self.config.get(f'plugins.{PLUGIN_NAME}.enabled'))
+        self.assertTrue(plugins.is_authorized(PLUGIN_NAME))
+        # the new file keeps its name, the old one is removed
+        self.assertEqual([f'{PLUGIN_NAME}-v2.zip'], os.listdir(self.plugins_dir))
+
+    def test_upgrade_takes_effect_after_restart(self):
+        make_plugin_zip(self.zip_path, secret='v1')
+        plugins = self._start_plugins()
+        self._authorize(plugins)
+        p = plugins.load_plugin_by_name(PLUGIN_NAME)
+        self.assertEqual('v1', sys.modules[p.__module__].SECRET)
+        plugins.upgrade_external_plugin(self._download_new_version('v2'), self.privkey)
+        # the old code keeps running from memory
+        self.assertIs(p, plugins.load_plugin_by_name(PLUGIN_NAME))
+        self.assertEqual(b'v1', plugins.read_file(PLUGIN_NAME, 'icon.txt'))
+        # as if electrum had been restarted
+        self._stop_plugins()
+        plugins = self._start_plugins()
+        self.assertTrue(plugins.is_authorized(PLUGIN_NAME))
+        p = plugins.load_plugin_by_name(PLUGIN_NAME)
+        self.assertEqual('v2', sys.modules[p.__module__].SECRET)
+
+    def test_upgrade_refuses_bytes_other_than_those_shown(self):
+        make_plugin_zip(self.zip_path, secret='v1')
+        plugins = self._start_plugins()
+        self._authorize(plugins)
+        manifest = self._download_new_version('v2')
+        make_plugin_zip(manifest['path'], secret='evil')
+        with self.assertRaises(IncorrectPluginHash):
+            plugins.upgrade_external_plugin(manifest, self.privkey)
+        self.assertTrue(plugins.is_authorized(PLUGIN_NAME))

### tests/test_zip_importer.py
@@ -0,0 +1,93 @@
+import importlib
+import importlib.util
+import json
+import os
+import sys
+import zipfile
+
+from electrum.zip_importer import MemoryZipImporter
+
+from . import ElectrumTestCase
+
+
+PLUGIN_NAME = 'toctou_test'
+
+
+def make_plugin_zip(path: str, *, secret: str = 'benign', name: str = PLUGIN_NAME) -> bytes:
+    """Writes a minimal, loadable cmdline plugin to `path`, and returns its bytes."""
+    with zipfile.ZipFile(path, 'w') as z:
+        z.writestr(f'{name}/manifest.json', json.dumps({
+            'name': name,
+            'fullname': 'TOCTOU test plugin',
+            'description': 'test fixture',
+            'available_for': ['cmdline'],
+        }))
+        z.writestr(f'{name}/__init__.py', f'SECRET = {secret!r}\n')
+        z.writestr(f'{name}/cmdline.py', '\n'.join([
+            'from electrum.plugin import BasePlugin',
+            f'SECRET = {secret!r}',
+            'class Plugin(BasePlugin):',
+            '    pass',
+            '',
+        ]))
+        z.writestr(f'{name}/icon.txt', secret)
+        z.writestr(f'{name}/sub/__init__.py', '')
+        z.writestr(f'{name}/sub/deep.py', f'DEEP = {secret!r}\n')
+    with open(path, 'rb') as f:
+        return f.read()
+
+
+class MemoryZipImporterTestCase(ElectrumTestCase):
+    """Tests for the in-memory archive itself, independent of the Plugins object."""
+
+    def setUp(self):
+        super().setUp()
+        self.path = os.path.join(self.electrum_path, 'p.zip')
+        self.blob = make_plugin_zip(self.path, secret='benign')
+
+    def _importer(self, blob=None) -> MemoryZipImporter:
+        return MemoryZipImporter(blob if blob is not None else self.blob,
+                         root_name='test_pkg_for_pluginzip', prefix=PLUGIN_NAME,
+                         archive_path=self.path)
+
+    def test_module_map(self):
+        b = self._importer()
+        self.assertTrue(b.is_package('test_pkg_for_pluginzip'))
+        self.assertFalse(b.is_package('test_pkg_for_pluginzip.cmdline'))
+        self.assertTrue(b.is_package('test_pkg_for_pluginzip.sub'))
+        self.assertIsNotNone(b.find_spec('test_pkg_for_pluginzip.sub.deep'))
+        self.assertIsNone(b.find_spec('test_pkg_for_pluginzip.nope'))
+        self.assertIsNone(b.find_spec('os'))  # never claims foreign names
+
+    def test_submodule_search_locations_is_empty(self):
+        # a non-empty entry would be a path, and PathFinder resolves a relative
+        # one against the process CWD, so there is no safe sentinel string
+        spec = self._importer().find_spec('test_pkg_for_pluginzip')
+        self.assertEqual([], spec.submodule_search_locations)
+
+    def test_importlib_cannot_resolve_submodule(self):
+        # check that importlib, which would read from disk,
+        # is disabled because search_locations is empty
+        importer = self._importer()
+        spec = importer.find_spec('test_pkg_for_pluginzip')
+        module = importlib.util.module_from_spec(spec)
+        sys.modules['test_pkg_for_pluginzip'] = module
+        self.addCleanup(sys.modules.pop, 'test_pkg_for_pluginzip', None)
+        with self.assertRaises(ModuleNotFoundError):
+            importlib.import_module('test_pkg_for_pluginzip.sub')
+        # find_spec works
+        spec2 = importer.find_spec('test_pkg_for_pluginzip.sub')
+        module2 = importlib.util.module_from_spec(spec2)
+
+    def test_read_resource(self):
+        self.assertEqual(b'benign', self._importer().read('icon.txt'))
+        with self.assertRaises(FileNotFoundError):
+            self._importer().read('no-such-file')
+
+    def test_corrupted_member_is_rejected(self):
+        blob = bytearray(self.blob)
+        with zipfile.ZipFile(self.path) as z:
+            offset = z.getinfo(f'{PLUGIN_NAME}/cmdline.py').header_offset
+        blob[offset + 60] ^= 0xff  # flip a bit inside the member payload
+        with self.assertRaises(zipfile.BadZipFile):
+            self._importer(bytes(blob)).read('cmdline.py')