MM-53355: install transitionally prepackaged plugins to filestore (#24225)

* move plugin signature verification to caller

The semantics for when plugin signature validation is required are unique to the caller, so move this logic there instead of masking it, thus simplifying some of the downstream code.

* support transitionally prepacked plugins

Transitionally prepackaged plugins are prepackaged plugins slated for unpackaging in some future release. Like prepackaged plugins, they automatically install or upgrade if the server is configured to enable that plugin, but unlike prepackaged plugins they don't add to the marketplace to allow for offline installs. In fact, if unlisted from the marketplace and not already enabled via `config.json`, a transitionally prepackaged plugin is essentially hidden.

To ensure a smooth transition in the future release when this plugin is no longer prepackaged at all, transitionally prepackaged plugins are persisted to the filestore as if they had been installed by the enduser. On the next restart, even while the plugin is still transitionally prepackaged, the version in the filestore will take priority. It remains possible for a transitionally prepackaged plugin to upgrade (and once again persist) if we ship a newer version before dropping it altogether.

Some complexity arises in a multi-server cluster, primarily because we don't want to deal with multiple servers writing the same object to the filestore. This is probably fine for S3, but has undefined semantics for regular filesystems, especially with some customers backing their files on any number of different fileshare technologies. To simplify the complexity, only the cluster leader persists transitionally prepackaged plugins.

Unfortunately, this too is complicated, since on upgrade to the first version with the transitionally prepackaged plugin, there is no guarantee that server will be the leader. In fact, as all nodes restart, there is no guarantee that any newly started server will start as the leader. So the persistence has to happen in a job-like fashion. The migration system might work, except we want the ability to run this repeatedly as we add to (or update) these transitionally prepackaged plugins. We also want to minimize the overhead required from the server to juggle any of this.

As a consequence, the persistence of transitionally prepackaged plugins occurs on every cluster leader change. Each server will try at most once to persist its collection of transitionally prepackaged plugins, and newly started servers will see the plugins in the filestore and skip this step altogether.

The current set of transitionally prepackaged plugins include the following, but this is expected to change:
* focalboard

* complete list of transitionally prepackaged plugins

* update plugin_install.go docs

* updated test plugins

* unit test transitionally prepackged plugins

* try restoring original working directory

* Apply suggestions from code review

Co-authored-by: Michael Kochell <6913320+mickmister@users.noreply.github.com>

* clarify processPrepackagedPlugins comment

---------

Co-authored-by: Michael Kochell <6913320+mickmister@users.noreply.github.com>

server/channels/app/plugin.go

@@ -4,6 +4,7 @@
 package app
 
 import (
+	"bytes"
 	"encoding/base64"
 	"fmt"
 	"io"
@@ -27,7 +28,6 @@ import (
 	"github.com/mattermost/mattermost/server/v8/channels/product"
 	"github.com/mattermost/mattermost/server/v8/channels/utils/fileutils"
 	"github.com/mattermost/mattermost/server/v8/platform/services/marketplace"
-	"github.com/mattermost/mattermost/server/v8/platform/shared/filestore"
 )
 
 // prepackagedPluginsDir is the hard-coded folder name where prepackaged plugins are bundled
@@ -253,13 +253,13 @@ func (ch *Channels) initPlugins(c *request.Context, pluginDir, webappPluginDir s
 		ch.srv.Log().Error("Failed to sync plugins from the file store", mlog.Err(err))
 	}
 
-	plugins := ch.processPrepackagedPlugins(prepackagedPluginsDir)
-	pluginsEnvironment = ch.GetPluginsEnvironment()
-	if pluginsEnvironment == nil {
-		ch.srv.Log().Info("Plugins environment not found, server is likely shutting down")
-		return
+	if err := ch.processPrepackagedPlugins(prepackagedPluginsDir); err != nil {
+		ch.srv.Log().Error("Failed to process prepackaged plugins", mlog.Err(err))
 	}
-	pluginsEnvironment.SetPrepackagedPlugins(plugins)
+	ch.pluginClusterLeaderListenerID = ch.srv.AddClusterLeaderChangedListener(func() {
+		ch.persistTransitionallyPrepackagedPlugins()
+	})
+	ch.persistTransitionallyPrepackagedPlugins()
 
 	// Sync plugin active state when config changes. Also notify plugins.
 	ch.pluginsLock.Lock()
@@ -352,18 +352,22 @@ func (ch *Channels) syncPlugins() *model.AppError {
 			}
 			defer bundle.Close()
 
-			var signature filestore.ReadCloseSeeker
 			if *ch.cfgSvc.Config().PluginSettings.RequirePluginSignature {
-				signature, appErr = ch.srv.fileReader(plugin.signaturePath)
+				signature, appErr := ch.srv.fileReader(plugin.signaturePath)
 				if appErr != nil {
 					logger.Error("Failed to open plugin signature from file store.", mlog.Err(appErr))
 					return
 				}
 				defer signature.Close()
+
+				if appErr = ch.verifyPlugin(bundle, signature); appErr != nil {
+					logger.Error("Failed to validate plugin signature", mlog.Err(appErr))
+					return
+				}
 			}
 
 			logger.Info("Syncing plugin from file store")
-			if _, err := ch.installPluginLocally(bundle, signature, installPluginLocallyAlways); err != nil && err.Id != "app.plugin.skip_installation.app_error" {
+			if _, err := ch.installPluginLocally(bundle, installPluginLocallyAlways); err != nil && err.Id != "app.plugin.skip_installation.app_error" {
 				logger.Error("Failed to sync plugin from file store", mlog.Err(err))
 			}
 		}(plugin)
@@ -388,6 +392,8 @@ func (ch *Channels) ShutDownPlugins() {
 
 	ch.RemoveConfigListener(ch.pluginConfigListenerID)
 	ch.pluginConfigListenerID = ""
+	ch.srv.RemoveClusterLeaderChangedListener(ch.pluginClusterLeaderListenerID)
+	ch.pluginClusterLeaderListenerID = ""
 
 	// Acquiring lock manually before cleaning up PluginsEnvironment.
 	ch.pluginsLock.Lock()
@@ -911,26 +917,48 @@ func (ch *Channels) getPluginsFromFilePaths(fileStorePaths []string) map[string]
 	return pluginSignaturePathMap
 }
 
-func (ch *Channels) processPrepackagedPlugins(pluginsDir string) []*plugin.PrepackagedPlugin {
-	prepackagedPluginsDir, found := fileutils.FindDir(pluginsDir)
+// processPrepackagedPlugins processes the plugins prepackaged with this server in the
+// prepackaged_plugins directory.
+//
+// If enabled, prepackaged plugins are installed or upgraded locally. A list of transitionally
+// prepackaged plugins is also collected for later persistence to the filestore.
+func (ch *Channels) processPrepackagedPlugins(prepackagedPluginsDir string) error {
+	prepackagedPluginsPath, found := fileutils.FindDir(prepackagedPluginsDir)
 	if !found {
 		ch.srv.Log().Debug("No prepackaged plugins directory found")
 		return nil
 	}
 
+	ch.srv.Log().Debug("Processing prepackaged plugins in directory", mlog.String("path", prepackagedPluginsPath))
+
 	var fileStorePaths []string
-	err := filepath.Walk(prepackagedPluginsDir, func(walkPath string, info os.FileInfo, err error) error {
+	err := filepath.Walk(prepackagedPluginsPath, func(walkPath string, info os.FileInfo, err error) error {
 		fileStorePaths = append(fileStorePaths, walkPath)
 		return nil
 	})
 	if err != nil {
-		ch.srv.Log().Error("Failed to walk prepackaged plugins", mlog.Err(err))
-		return nil
+		return errors.Wrap(err, "failed to walk prepackaged plugins")
 	}
 
 	pluginSignaturePathMap := ch.getPluginsFromFilePaths(fileStorePaths)
-	plugins := make([]*plugin.PrepackagedPlugin, 0, len(pluginSignaturePathMap))
-	prepackagedPlugins := make(chan *plugin.PrepackagedPlugin, len(pluginSignaturePathMap))
+	plugins := make(chan *plugin.PrepackagedPlugin, len(pluginSignaturePathMap))
+
+	// Before processing any prepackaged plugins, take a snapshot of the available manifests
+	// to decide what was synced from the filestore.
+	pluginsEnvironment := ch.GetPluginsEnvironment()
+	if pluginsEnvironment == nil {
+		return errors.New("pluginsEnvironment is nil")
+	}
+
+	availablePlugins, err := pluginsEnvironment.Available()
+	if err != nil {
+		return errors.Wrap(err, "failed to list available plugins")
+	}
+
+	availablePluginsMap := make(map[string]*model.BundleInfo, len(availablePlugins))
+	for _, bundleInfo := range availablePlugins {
+		availablePluginsMap[bundleInfo.Manifest.Id] = bundleInfo
+	}
 
 	var wg sync.WaitGroup
 	for _, psPath := range pluginSignaturePathMap {
@@ -946,18 +974,29 @@ func (ch *Channels) processPrepackagedPlugins(pluginsDir string) []*plugin.Prepa
 				ch.srv.Log().Error("Failed to install prepackaged plugin", mlog.String("bundle_path", psPath.bundlePath), mlog.Err(err))
 				return
 			}
-			prepackagedPlugins <- p
+
+			plugins <- p
 		}(psPath)
 	}
 
 	wg.Wait()
-	close(prepackagedPlugins)
+	close(plugins)
 
-	for p := range prepackagedPlugins {
-		plugins = append(plugins, p)
+	prepackagedPlugins := make([]*plugin.PrepackagedPlugin, 0, len(pluginSignaturePathMap))
+	transitionallyPrepackagedPlugins := make([]*plugin.PrepackagedPlugin, 0)
+	for p := range plugins {
+		if ch.pluginIsTransitionallyPrepackaged(p.Manifest.Id) {
+			if ch.shouldPersistTransitionallyPrepackagedPlugin(availablePluginsMap, p) {
+				transitionallyPrepackagedPlugins = append(transitionallyPrepackagedPlugins, p)
+			}
+		} else {
+			prepackagedPlugins = append(prepackagedPlugins, p)
+		}
 	}
 
-	return plugins
+	pluginsEnvironment.SetPrepackagedPlugins(prepackagedPlugins, transitionallyPrepackagedPlugins)
+
+	return nil
 }
 
 // processPrepackagedPlugin will return the prepackaged plugin metadata and will also
@@ -984,25 +1023,172 @@ func (ch *Channels) processPrepackagedPlugin(pluginPath *pluginSignaturePath) (*
 		return nil, errors.Wrapf(err, "Failed to get prepackaged plugin %s", pluginPath.bundlePath)
 	}
 
+	logger = logger.With(mlog.String("plugin_id", plugin.Manifest.Id))
+
 	// Skip installing the plugin at all if automatic prepackaged plugins is disabled
 	if !*ch.cfgSvc.Config().PluginSettings.AutomaticPrepackagedPlugins {
+		logger.Info("Not installing prepackaged plugin: automatic prepackaged plugins disabled")
 		return plugin, nil
 	}
 
 	// Skip installing if the plugin is has not been previously enabled.
 	pluginState := ch.cfgSvc.Config().PluginSettings.PluginStates[plugin.Manifest.Id]
 	if pluginState == nil || !pluginState.Enable {
+		logger.Info("Not installing prepackaged plugin: not previously enabled")
 		return plugin, nil
 	}
 
-	logger.Info("Installing prepackaged plugin")
-	if _, err := ch.installExtractedPlugin(plugin.Manifest, pluginDir, installPluginLocallyOnlyIfNewOrUpgrade); err != nil {
+	if _, err := ch.installExtractedPlugin(plugin.Manifest, pluginDir, installPluginLocallyOnlyIfNewOrUpgrade); err != nil && err.Id != "app.plugin.skip_installation.app_error" {
 		return nil, errors.Wrapf(err, "Failed to install extracted prepackaged plugin %s", pluginPath.bundlePath)
 	}
 
 	return plugin, nil
 }
 
+var transitionallyPrepackagedPlugins = []string{
+	"antivirus",
+	"focalboard",
+	"mattermost-autolink",
+	"com.mattermost.aws-sns",
+	"com.mattermost.plugin-channel-export",
+	"com.mattermost.confluence",
+	"com.mattermost.custom-attributes",
+	"jenkins",
+	"jitsi",
+	"com.mattermost.plugin-todo",
+	"com.mattermost.welcomebot",
+	"com.mattermost.apps",
+}
+
+// pluginIsTransitionallyPrepackaged identifies plugin ids that are currently prepackaged but
+// slated for future removal.
+func (ch *Channels) pluginIsTransitionallyPrepackaged(pluginID string) bool {
+	for _, id := range transitionallyPrepackagedPlugins {
+		if id == pluginID {
+			return true
+		}
+	}
+
+	return false
+}
+
+// shouldPersistTransitionallyPrepackagedPlugin determines if a transitionally prepackaged plugin
+// should be persisted to the filestore, taking into account whether it's already enabled and
+// would improve on what's already in the filestore.
+func (ch *Channels) shouldPersistTransitionallyPrepackagedPlugin(availablePluginsMap map[string]*model.BundleInfo, p *plugin.PrepackagedPlugin) bool {
+	logger := ch.srv.Log().With(mlog.String("plugin_id", p.Manifest.Id), mlog.String("prepackaged_version", p.Manifest.Version))
+
+	// Ignore the plugin altogether unless it was previously enabled.
+	pluginState := ch.cfgSvc.Config().PluginSettings.PluginStates[p.Manifest.Id]
+	if pluginState == nil || !pluginState.Enable {
+		logger.Debug("Should not persist transitionally prepackaged plugin: not previously enabled")
+		return false
+	}
+
+	// Ignore the plugin if the same or newer version is already available
+	// (having previously synced from the filestore).
+	existing, found := availablePluginsMap[p.Manifest.Id]
+	if !found {
+		logger.Info("Should persist transitionally prepackaged plugin: not currently in filestore")
+		return true
+	}
+
+	prepackagedVersion, err := semver.Parse(p.Manifest.Version)
+	if err != nil {
+		logger.Error("Should not persist transitionally prepackged plugin: invalid prepackaged version", mlog.Err(err))
+		return false
+	}
+
+	logger = logger.With(mlog.String("existing_version", existing.Manifest.Version))
+
+	existingVersion, err := semver.Parse(existing.Manifest.Version)
+	if err != nil {
+		// Consider this an old version and replace with the prepackaged version instead.
+		logger.Warn("Should persist transitionally prepackged plugin: invalid existing version", mlog.Err(err))
+		return true
+	}
+
+	if prepackagedVersion.GT(existingVersion) {
+		logger.Info("Should persist transitionally prepackged plugin: newer version")
+		return true
+	}
+
+	logger.Info("Should not persist transitionally prepackged plugin: not a newer version")
+	return false
+}
+
+// persistTransitionallyPrepackagedPlugins writes plugins that are transitionally prepackaged with
+// the server to the filestore to allow their continued use when the plugin eventually stops being
+// prepackaged.
+//
+// We identify which plugins need to be persisted during startup via processPrepackagedPlugins.
+// Once we persist the set of plugins to the filestore, we clear the list to prevent this server
+// from trying again.
+//
+// In a multi-server cluster, only the cluster leader should persist these plugins to avoid
+// concurrent writes to the filestore. But during an upgrade, there's no guarantee that a freshly
+// upgraded server will be the cluster leader to perform this step in a timely fashion, so the
+// persistence has to be able to happen sometime after startup. Additionally, while this is a
+// kind of migration, it's not a one off: new versions of these plugins may still be shipped
+// during the transition period, or new plugins may be added to the list.
+//
+// So instead of a one-time migration, we opt to run this method every time the cluster leader
+// changes, but minimizing rework. More than one server may end up persisting the same plugin
+// (but never concurrently!), but all servers will eventually converge on this method becoming a
+// no-op (until this set of plugins changes in a subsequent release).
+//
+// Finally, if an error occurs persisting the plugin, we don't try again until the server restarts,
+// or another server becomes cluster leader.
+func (ch *Channels) persistTransitionallyPrepackagedPlugins() {
+	if !ch.srv.IsLeader() {
+		ch.srv.Log().Debug("Not persisting transitionally prepackaged plugins: not the leader")
+		return
+	}
+
+	pluginsEnvironment := ch.GetPluginsEnvironment()
+	if pluginsEnvironment == nil {
+		ch.srv.Log().Debug("Not persisting transitionally prepackaged plugins: no plugin environment")
+		return
+	}
+
+	transitionallyPrepackagedPlugins := pluginsEnvironment.TransitionallyPrepackagedPlugins()
+	if len(transitionallyPrepackagedPlugins) == 0 {
+		ch.srv.Log().Debug("Not persisting transitionally prepackaged plugins: none found")
+		return
+	}
+
+	var wg sync.WaitGroup
+	for _, p := range transitionallyPrepackagedPlugins {
+		wg.Add(1)
+		go func(p *plugin.PrepackagedPlugin) {
+			defer wg.Done()
+
+			logger := ch.srv.Log().With(mlog.String("plugin_id", p.Manifest.Id), mlog.String("version", p.Manifest.Version))
+
+			logger.Info("Persisting transitionally prepackaged plugin")
+
+			bundleReader, err := os.Open(p.Path)
+			if err != nil {
+				logger.Error("Failed to read transitionally prepackaged plugin", mlog.Err(err))
+			}
+			defer bundleReader.Close()
+
+			signatureReader := bytes.NewReader(p.Signature)
+
+			// Write the plugin to the filestore, but don't bother notifying the peers,
+			// as there's no reason to reload the plugin to run the same version again.
+			appErr := ch.installPluginToFilestore(p.Manifest, bundleReader, signatureReader)
+			if appErr != nil {
+				logger.Error("Failed to persist transitionally prepackaged plugin", mlog.Err(err))
+			}
+		}(p)
+	}
+	wg.Wait()
+
+	pluginsEnvironment.ClearTransitionallyPrepackagedPlugins()
+	ch.srv.Log().Info("Finished persisting transitionally prepackaged plugins")
+}
+
 // buildPrepackagedPlugin builds a PrepackagedPlugin from the plugin at the given path, additionally returning the directory in which it was extracted.
 func (ch *Channels) buildPrepackagedPlugin(pluginPath *pluginSignaturePath, pluginFile io.ReadSeeker, tmpDir string) (*plugin.PrepackagedPlugin, string, error) {
 	manifest, pluginDir, appErr := extractPlugin(pluginFile, tmpDir)