Adding interplugin communication. (#12829)

* Adding interplugin communication.

* Naming changes and moving ResponseTransfer to own file.

* Fix.

* Tests and moving to buffering bytes.

* Switching API to passing plugin ID through path rather than a header.

* Review feedback.

app/plugin_api.go

@@ -10,6 +10,7 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"net/url"
 	"path/filepath"
 	"strings"
 
@@ -825,3 +826,29 @@ func (api *PluginAPI) DeleteBotIconImage(userId string) *model.AppError {
 
 	return api.app.DeleteBotIconImage(userId)
 }
+
+func (api *PluginAPI) PluginHTTP(request *http.Request) *http.Response {
+	split := strings.SplitN(request.URL.Path, "/", 3)
+	if len(split) != 3 {
+		return &http.Response{
+			StatusCode: http.StatusBadRequest,
+			Body:       ioutil.NopCloser(bytes.NewBufferString("Not enough URL. Form of URL should be /<pluginid>/*")),
+		}
+	}
+	destinationPluginId := split[1]
+	newURL, err := url.Parse("/" + split[2])
+	request.URL = newURL
+	if destinationPluginId == "" || err != nil {
+		message := "No plugin specified. Form of URL should be /<pluginid>/*"
+		if err != nil {
+			message = "Form of URL should be /<pluginid>/* Error: " + err.Error()
+		}
+		return &http.Response{
+			StatusCode: http.StatusBadRequest,
+			Body:       ioutil.NopCloser(bytes.NewBufferString(message)),
+		}
+	}
+	responseTransfer := &PluginResponseWriter{}
+	api.app.ServeInterPluginRequest(responseTransfer, request, api.id, destinationPluginId)
+	return responseTransfer.GenerateResponse()
+}

app/plugin_api_test.go

@@ -26,7 +26,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func setupPluginApiTest(t *testing.T, pluginCode string, pluginManifest string, pluginId string, app *App) string {
+func setupMultiPluginApiTest(t *testing.T, pluginCodes []string, pluginManifests []string, pluginIds []string, app *App) string {
 	pluginDir, err := ioutil.TempDir("", "")
 	require.NoError(t, err)
 	webappPluginDir, err := ioutil.TempDir("", "")
@@ -37,20 +37,29 @@ func setupPluginApiTest(t *testing.T, pluginCode string, pluginManifest string,
 	env, err := plugin.NewEnvironment(app.NewPluginAPI, pluginDir, webappPluginDir, app.Log)
 	require.NoError(t, err)
 
-	backend := filepath.Join(pluginDir, pluginId, "backend.exe")
-	utils.CompileGo(t, pluginCode, backend)
+	require.Equal(t, len(pluginCodes), len(pluginIds))
+	require.Equal(t, len(pluginManifests), len(pluginIds))
 
-	ioutil.WriteFile(filepath.Join(pluginDir, pluginId, "plugin.json"), []byte(pluginManifest), 0600)
-	manifest, activated, reterr := env.Activate(pluginId)
-	require.Nil(t, reterr)
-	require.NotNil(t, manifest)
-	require.True(t, activated)
+	for i, pluginId := range pluginIds {
+		backend := filepath.Join(pluginDir, pluginId, "backend.exe")
+		utils.CompileGo(t, pluginCodes[i], backend)
+
+		ioutil.WriteFile(filepath.Join(pluginDir, pluginId, "plugin.json"), []byte(pluginManifests[i]), 0600)
+		manifest, activated, reterr := env.Activate(pluginId)
+		require.Nil(t, reterr)
+		require.NotNil(t, manifest)
+		require.True(t, activated)
+	}
 
 	app.SetPluginsEnvironment(env)
 
 	return pluginDir
 }
 
+func setupPluginApiTest(t *testing.T, pluginCode string, pluginManifest string, pluginId string, app *App) string {
+	return setupMultiPluginApiTest(t, []string{pluginCode}, []string{pluginManifest}, []string{pluginId}, app)
+}
+
 func TestPublicFilesPathConfiguration(t *testing.T) {
 	th := Setup(t).InitBasic()
 	defer th.TearDown()
@@ -1462,3 +1471,98 @@ func TestPluginAddUserToChannel(t *testing.T) {
 	require.Equal(t, th.BasicChannel.Id, member.ChannelId)
 	require.Equal(t, th.BasicUser.Id, member.UserId)
 }
+
+func TestInterpluginPluginHTTP(t *testing.T) {
+	th := Setup(t).InitBasic()
+	defer th.TearDown()
+
+	setupMultiPluginApiTest(t,
+		[]string{`
+		package main
+
+		import (
+			"github.com/mattermost/mattermost-server/plugin"
+			"bytes"
+			"net/http"
+		)
+
+		type MyPlugin struct {
+			plugin.MattermostPlugin
+		}
+
+		func (p *MyPlugin) ServeHTTP(c *plugin.Context, w http.ResponseWriter, r *http.Request) {
+			if r.URL.Path != "/api/v2/test" {
+				return
+			}
+			buf := bytes.Buffer{}
+			buf.ReadFrom(r.Body)
+			resp := "we got:" + buf.String()
+			w.WriteHeader(598)
+			w.Write([]byte(resp))
+		}
+
+		func main() {
+			plugin.ClientMain(&MyPlugin{})
+		}
+		`,
+			`
+		package main
+
+		import (
+			"github.com/mattermost/mattermost-server/plugin"
+			"github.com/mattermost/mattermost-server/model"
+			"bytes"
+			"net/http"
+			"io/ioutil"
+		)
+
+		type MyPlugin struct {
+			plugin.MattermostPlugin
+		}
+
+		func (p *MyPlugin) MessageWillBePosted(c *plugin.Context, post *model.Post) (*model.Post, string) {
+			buf := bytes.Buffer{}
+			buf.WriteString("This is the request")
+			req, err := http.NewRequest("GET", "/testplugininterserver/api/v2/test", &buf)
+			if err != nil {
+				return nil, err.Error()
+			}
+			req.Header.Add("Mattermost-User-Id", "userid")
+			resp := p.API.PluginHTTP(req)
+			if resp == nil {
+				return nil, "Nil resp"
+			}
+			if resp.Body == nil {
+				return nil, "Nil body"
+			}
+			respbody, err := ioutil.ReadAll(resp.Body)
+			if err != nil {
+				return nil, err.Error()
+			}
+			if resp.StatusCode != 598 {
+				return nil, "wrong status " + string(respbody)
+			}
+			return nil, string(respbody)
+		}
+
+		func main() {
+			plugin.ClientMain(&MyPlugin{})
+		}
+		`,
+		},
+		[]string{
+			`{"id": "testplugininterserver", "backend": {"executable": "backend.exe"}}`,
+			`{"id": "testplugininterclient", "backend": {"executable": "backend.exe"}}`,
+		},
+		[]string{
+			"testplugininterserver",
+			"testplugininterclient",
+		},
+		th.App,
+	)
+
+	hooks, err := th.App.GetPluginsEnvironment().HooksForPlugin("testplugininterclient")
+	require.NoError(t, err)
+	_, ret := hooks.MessageWillBePosted(nil, nil)
+	assert.Equal(t, "we got:This is the request", ret)
+}

app/plugin_requests.go

@@ -42,6 +42,37 @@ func (a *App) ServePluginRequest(w http.ResponseWriter, r *http.Request) {
 	a.servePluginRequest(w, r, hooks.ServeHTTP)
 }
 
+func (a *App) ServeInterPluginRequest(w http.ResponseWriter, r *http.Request, sourcePluginId, destinationPluginId string) {
+	pluginsEnvironment := a.GetPluginsEnvironment()
+	if pluginsEnvironment == nil {
+		err := model.NewAppError("ServeInterPluginRequest", "app.plugin.disabled.app_error", nil, "Plugin enviroment not found.", http.StatusNotImplemented)
+		a.Log.Error(err.Error())
+		w.WriteHeader(err.StatusCode)
+		w.Header().Set("Content-Type", "application/json")
+		w.Write([]byte(err.ToJson()))
+		return
+	}
+
+	hooks, err := pluginsEnvironment.HooksForPlugin(destinationPluginId)
+	if err != nil {
+		a.Log.Error("Access to route for non-existent plugin in inter plugin request",
+			mlog.String("sourse_plugin_id", sourcePluginId),
+			mlog.String("destination_plugin_id", destinationPluginId),
+			mlog.Err(err),
+		)
+		http.NotFound(w, r)
+		return
+	}
+
+	context := &plugin.Context{
+		RequestId:      model.NewId(),
+		UserAgent:      r.UserAgent(),
+		SourcePluginId: sourcePluginId,
+	}
+
+	hooks.ServeHTTP(context, w, r)
+}
+
 // ServePluginPublicRequest serves public plugin files
 // at the URL http(s)://$SITE_URL/plugins/$PLUGIN_ID/public/{anything}
 func (a *App) ServePluginPublicRequest(w http.ResponseWriter, r *http.Request) {

app/response_transfer.go

@@ -0,0 +1,70 @@
+// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved.
+// See License.txt for license information.
+
+package app
+
+import (
+	"bytes"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"strconv"
+	"strings"
+)
+
+type PluginResponseWriter struct {
+	bytes.Buffer
+	headers    http.Header
+	statusCode int
+}
+
+func (rt *PluginResponseWriter) Header() http.Header {
+	if rt.headers == nil {
+		rt.headers = make(http.Header)
+	}
+	return rt.headers
+}
+
+func (rt *PluginResponseWriter) WriteHeader(statusCode int) {
+	rt.statusCode = statusCode
+}
+
+// From net/http/httptest/recorder.go
+func parseContentLength(cl string) int64 {
+	cl = strings.TrimSpace(cl)
+	if cl == "" {
+		return -1
+	}
+	n, err := strconv.ParseInt(cl, 10, 64)
+	if err != nil {
+		return -1
+	}
+	return n
+
+}
+
+func (rt *PluginResponseWriter) GenerateResponse() *http.Response {
+	res := &http.Response{
+		Proto:      "HTTP/1.1",
+		ProtoMajor: 1,
+		ProtoMinor: 1,
+		StatusCode: rt.statusCode,
+		Header:     rt.headers.Clone(),
+	}
+
+	if res.StatusCode == 0 {
+		res.StatusCode = http.StatusOK
+	}
+
+	res.Status = fmt.Sprintf("%03d %s", res.StatusCode, http.StatusText(res.StatusCode))
+
+	if rt.Len() > 0 {
+		res.Body = ioutil.NopCloser(rt)
+	} else {
+		res.Body = http.NoBody
+	}
+
+	res.ContentLength = parseContentLength(rt.headers.Get("Content-Length"))
+
+	return res
+}

plugin/api.go

@@ -5,6 +5,7 @@ package plugin
 
 import (
 	"io"
+	"net/http"
 
 	plugin "github.com/hashicorp/go-plugin"
 	"github.com/mattermost/mattermost-server/model"
@@ -747,6 +748,11 @@ type API interface {
 	//
 	// Minimum server version: 5.14
 	DeleteBotIconImage(botUserId string) *model.AppError
+
+	// PluginHTTP allows inter-plugin requests to plugin APIs.
+	//
+	// Minimum server version: 5.18
+	PluginHTTP(request *http.Request) *http.Response
 }
 
 var handshake = plugin.HandshakeConfig{

plugin/client_rpc.go

@@ -56,11 +56,13 @@ func (p *hooksPlugin) Client(b *plugin.MuxBroker, client *rpc.Client) (interface
 }
 
 type apiRPCClient struct {
-	client *rpc.Client
+	client    *rpc.Client
+	muxBroker *plugin.MuxBroker
 }
 
 type apiRPCServer struct {
-	impl API
+	impl      API
+	muxBroker *plugin.MuxBroker
 }
 
 // ErrorString is a fallback for sending unregistered implementations of the error interface across
@@ -171,7 +173,8 @@ type Z_OnActivateReturns struct {
 func (g *hooksRPCClient) OnActivate() error {
 	muxId := g.muxBroker.NextId()
 	go g.muxBroker.AcceptAndServe(muxId, &apiRPCServer{
-		impl: g.apiImpl,
+		impl:      g.apiImpl,
+		muxBroker: g.muxBroker,
 	})
 
 	_args := &Z_OnActivateArgs{
@@ -192,7 +195,8 @@ func (s *hooksRPCServer) OnActivate(args *Z_OnActivateArgs, returns *Z_OnActivat
 	}
 
 	s.apiRPCClient = &apiRPCClient{
-		client: rpc.NewClient(connection),
+		client:    rpc.NewClient(connection),
+		muxBroker: s.muxBroker,
 	}
 
 	if mmplugin, ok := s.impl.(interface {
@@ -363,6 +367,76 @@ func (s *hooksRPCServer) ServeHTTP(args *Z_ServeHTTPArgs, returns *struct{}) err
 	return nil
 }
 
+type Z_PluginHTTPArgs struct {
+	Request     *http.Request
+	RequestBody []byte
+}
+
+type Z_PluginHTTPReturns struct {
+	Response     *http.Response
+	ResponseBody []byte
+}
+
+func (g *apiRPCClient) PluginHTTP(request *http.Request) *http.Response {
+	forwardedRequest := &http.Request{
+		Method:     request.Method,
+		URL:        request.URL,
+		Proto:      request.Proto,
+		ProtoMajor: request.ProtoMajor,
+		ProtoMinor: request.ProtoMinor,
+		Header:     request.Header,
+		Host:       request.Host,
+		RemoteAddr: request.RemoteAddr,
+		RequestURI: request.RequestURI,
+	}
+
+	requestBody, err := ioutil.ReadAll(request.Body)
+	if err != nil {
+		log.Printf("RPC call to PluginHTTP API failed: %s", err.Error())
+		return nil
+	}
+	request.Body.Close()
+	request.Body = nil
+
+	_args := &Z_PluginHTTPArgs{
+		Request:     forwardedRequest,
+		RequestBody: requestBody,
+	}
+
+	_returns := &Z_PluginHTTPReturns{}
+	if err := g.client.Call("Plugin.PluginHTTP", _args, _returns); err != nil {
+		log.Printf("RPC call to PluginHTTP API failed: %s", err.Error())
+		return nil
+	}
+
+	_returns.Response.Body = ioutil.NopCloser(bytes.NewBuffer(_returns.ResponseBody))
+
+	return _returns.Response
+}
+
+func (s *apiRPCServer) PluginHTTP(args *Z_PluginHTTPArgs, returns *Z_PluginHTTPReturns) error {
+	args.Request.Body = ioutil.NopCloser(bytes.NewBuffer(args.RequestBody))
+
+	if hook, ok := s.impl.(interface {
+		PluginHTTP(request *http.Request) *http.Response
+	}); ok {
+		response := hook.PluginHTTP(args.Request)
+
+		responseBody, err := ioutil.ReadAll(response.Body)
+		if err != nil {
+			return encodableError(fmt.Errorf("RPC call to PluginHTTP API failed: %s", err.Error()))
+		}
+		response.Body.Close()
+		response.Body = nil
+
+		returns.Response = response
+		returns.ResponseBody = responseBody
+	} else {
+		return encodableError(fmt.Errorf("API PluginHTTP called but not implemented."))
+	}
+	return nil
+}
+
 func init() {
 	hookNameToId["FileWillBeUploaded"] = FileWillBeUploadedId
 }

plugin/context.go

@@ -12,4 +12,5 @@ type Context struct {
 	IpAddress      string
 	AcceptLanguage string
 	UserAgent      string
+	SourcePluginId string
 }

plugin/interface_generator/main.go

@@ -392,17 +392,18 @@ func getPluginPackageDir() string {
 func removeExcluded(info *PluginInterfaceInfo) *PluginInterfaceInfo {
 	toBeExcluded := func(item string) bool {
 		excluded := []string{
-			"OnActivate",
+			"FileWillBeUploaded",
 			"Implemented",
 			"LoadPluginConfiguration",
-			"ServeHTTP",
-			"FileWillBeUploaded",
-			"MessageWillBePosted",
-			"MessageWillBeUpdated",
 			"LogDebug",
+			"LogError",
 			"LogInfo",
 			"LogWarn",
-			"LogError",
+			"MessageWillBePosted",
+			"MessageWillBeUpdated",
+			"OnActivate",
+			"PluginHTTP",
+			"ServeHTTP",
 		}
 		for _, exclusion := range excluded {
 			if exclusion == item {

plugin/plugintest/api.go

@@ -6,9 +6,11 @@ package plugintest
 
 import (
 	io "io"
+	http "net/http"
+
+	mock "github.com/stretchr/testify/mock"
 
 	model "github.com/mattermost/mattermost-server/model"
-	mock "github.com/stretchr/testify/mock"
 )
 
 // API is an autogenerated mock type for the API type
@@ -2334,6 +2336,22 @@ func (_m *API) PermanentDeleteBot(botUserId string) *model.AppError {
 	return r0
 }
 
+// PluginHTTP provides a mock function with given fields: request
+func (_m *API) PluginHTTP(request *http.Request) *http.Response {
+	ret := _m.Called(request)
+
+	var r0 *http.Response
+	if rf, ok := ret.Get(0).(func(*http.Request) *http.Response); ok {
+		r0 = rf(request)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*http.Response)
+		}
+	}
+
+	return r0
+}
+
 // PublishWebSocketEvent provides a mock function with given fields: event, payload, broadcast
 func (_m *API) PublishWebSocketEvent(event string, payload map[string]interface{}, broadcast *model.WebsocketBroadcast) {
 	_m.Called(event, payload, broadcast)