[MM-16473] Make plugins' ServerHTTP http.ResponseWriter hijackable (#14822)

* Make plugins' ServerHTTP http.ResponseWriter hijackable

* Rename brw to align with docs

* Fix error handling

app/plugin_api_test.go

@@ -1578,6 +1578,86 @@ func TestPluginAPIGetPostsForChannel(t *testing.T) {
 	require.Equal(expectedPosts, postList.ToSlice())
 }
 
+func TestPluginHTTPConnHijack(t *testing.T) {
+	th := Setup(t).InitBasic()
+	defer th.TearDown()
+
+	testFolder, found := fileutils.FindDir("mattermost-server/app/plugin_api_tests")
+	require.True(t, found, "Cannot find tests folder")
+	fullPath := path.Join(testFolder, "manual.test_http_hijack_plugin", "main.go")
+
+	pluginCode, err := ioutil.ReadFile(fullPath)
+	require.NoError(t, err)
+	require.NotEmpty(t, pluginCode)
+
+	tearDown, ids, errors := SetAppEnvironmentWithPlugins(t, []string{string(pluginCode)}, th.App, th.App.NewPluginAPI)
+	defer tearDown()
+	require.NoError(t, errors[0])
+	require.Len(t, ids, 1)
+
+	pluginID := ids[0]
+	require.NotEmpty(t, pluginID)
+
+	reqURL := fmt.Sprintf("http://localhost:%d/plugins/%s", th.Server.ListenAddr.Port, pluginID)
+	req, err := http.NewRequest("GET", reqURL, nil)
+	require.NoError(t, err)
+
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	require.NoError(t, err)
+
+	defer resp.Body.Close()
+
+	body, err := ioutil.ReadAll(resp.Body)
+	require.NoError(t, err)
+	require.Equal(t, "OK", string(body))
+}
+
+func TestPluginHTTPUpgradeWebSocket(t *testing.T) {
+	th := Setup(t).InitBasic()
+	defer th.TearDown()
+
+	testFolder, found := fileutils.FindDir("mattermost-server/app/plugin_api_tests")
+	require.True(t, found, "Cannot find tests folder")
+	fullPath := path.Join(testFolder, "manual.test_http_upgrade_websocket_plugin", "main.go")
+
+	pluginCode, err := ioutil.ReadFile(fullPath)
+	require.NoError(t, err)
+	require.NotEmpty(t, pluginCode)
+
+	tearDown, ids, errors := SetAppEnvironmentWithPlugins(t, []string{string(pluginCode)}, th.App, th.App.NewPluginAPI)
+	defer tearDown()
+	require.NoError(t, errors[0])
+	require.Len(t, ids, 1)
+
+	pluginID := ids[0]
+	require.NotEmpty(t, pluginID)
+
+	reqURL := fmt.Sprintf("ws://localhost:%d/plugins/%s", th.Server.ListenAddr.Port, pluginID)
+	wsc, err := model.NewWebSocketClient(reqURL, "")
+	require.Nil(t, err)
+	require.NotNil(t, wsc)
+
+	wsc.Listen()
+	defer wsc.Close()
+
+	resp := <-wsc.ResponseChannel
+	require.Equal(t, resp.Status, model.STATUS_OK)
+
+	for i := 0; i < 10; i++ {
+		wsc.SendMessage("custom_action", map[string]interface{}{"value": i})
+		var resp *model.WebSocketResponse
+		select {
+		case resp = <-wsc.ResponseChannel:
+		case <-time.After(1 * time.Second):
+		}
+		require.NotNil(t, resp)
+		require.Equal(t, resp.Status, model.STATUS_OK)
+		require.Equal(t, "custom_action", resp.Data["action"])
+		require.Equal(t, float64(i), resp.Data["value"])
+	}
+}
+
 func TestPluginAPISearchPostsInTeamByUser(t *testing.T) {
 	th := Setup(t).InitBasic()
 	defer th.TearDown()

app/plugin_api_tests/manual.test_http_hijack_plugin/main.go

@@ -0,0 +1,35 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package main
+
+import (
+	"net/http"
+
+	"github.com/mattermost/mattermost-server/v5/plugin"
+)
+
+type Plugin struct {
+	plugin.MattermostPlugin
+}
+
+func (p *Plugin) ServeHTTP(c *plugin.Context, w http.ResponseWriter, r *http.Request) {
+	hj, ok := w.(http.Hijacker)
+	if !ok {
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	conn, brw, err := hj.Hijack()
+	if conn == nil || brw == nil || err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	conn.Write([]byte("HTTP/1.1 200\n\nOK"))
+	conn.Close()
+}
+
+func main() {
+	plugin.ClientMain(&Plugin{})
+}

app/plugin_api_tests/manual.test_http_upgrade_websocket_plugin/main.go

@@ -0,0 +1,45 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package main
+
+import (
+	"bytes"
+	"net/http"
+
+	"github.com/mattermost/mattermost-server/v5/model"
+	"github.com/mattermost/mattermost-server/v5/plugin"
+
+	"github.com/gorilla/websocket"
+)
+
+type Plugin struct {
+	plugin.MattermostPlugin
+}
+
+func (p *Plugin) ServeHTTP(c *plugin.Context, w http.ResponseWriter, r *http.Request) {
+	upgrader := websocket.Upgrader{}
+
+	ws, err := upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+	defer ws.Close()
+
+	for {
+		mt, msg, err := ws.ReadMessage()
+		if err != nil {
+			break
+		}
+		req := model.WebSocketRequestFromJson(bytes.NewReader(msg))
+		resp := model.NewWebSocketResponse("OK", req.Seq, map[string]interface{}{"action": req.Action, "value": req.Data["value"]})
+		if err = ws.WriteMessage(mt, []byte(resp.ToJson())); err != nil {
+			break
+		}
+	}
+}
+
+func main() {
+	plugin.ClientMain(&Plugin{})
+}

plugin/hijack.go

@@ -0,0 +1,205 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package plugin
+
+import (
+	"bufio"
+	"errors"
+	"net"
+	"net/http"
+	"net/rpc"
+	"time"
+)
+
+const (
+	hijackedConnReadBufSize = 4096
+)
+
+var (
+	ErrNotHijacked     = errors.New("response is not hijacked")
+	ErrAlreadyHijacked = errors.New("response was already hijacked")
+	ErrCannotHijack    = errors.New("response cannot be hijacked")
+)
+
+func (w *httpResponseWriterRPCServer) HjConnRWRead(b []byte, reply *[]byte) error {
+	if w.hjr == nil {
+		return ErrNotHijacked
+	}
+	data := make([]byte, len(b))
+	n, err := w.hjr.bufrw.Read(data)
+	if err != nil {
+		return err
+	}
+	*reply = data[:n]
+	return nil
+}
+
+func (w *httpResponseWriterRPCServer) HjConnRWWrite(b []byte, reply *int) error {
+	if w.hjr == nil {
+		return ErrNotHijacked
+	}
+	n, err := w.hjr.bufrw.Write(b)
+	if err != nil {
+		return err
+	}
+	*reply = n
+	return nil
+}
+
+func (w *httpResponseWriterRPCServer) HjConnRead(size int, reply *[]byte) error {
+	if w.hjr == nil {
+		return ErrNotHijacked
+	}
+	if len(w.hjr.readBuf) < size {
+		w.hjr.readBuf = make([]byte, size)
+	}
+	n, err := w.hjr.conn.Read(w.hjr.readBuf[:size])
+	if err != nil {
+		return err
+	}
+	*reply = w.hjr.readBuf[:n]
+	return nil
+}
+
+func (w *httpResponseWriterRPCServer) HjConnWrite(b []byte, reply *int) error {
+	if w.hjr == nil {
+		return ErrNotHijacked
+	}
+	n, err := w.hjr.conn.Write(b)
+	if err != nil {
+		return err
+	}
+	*reply = n
+	return nil
+}
+
+func (w *httpResponseWriterRPCServer) HjConnClose(args struct{}, reply *struct{}) error {
+	if w.hjr == nil {
+		return ErrNotHijacked
+	}
+	return w.hjr.conn.Close()
+}
+
+func (w *httpResponseWriterRPCServer) HjConnSetDeadline(t time.Time, reply *struct{}) error {
+	if w.hjr == nil {
+		return ErrNotHijacked
+	}
+	return w.hjr.conn.SetDeadline(t)
+}
+
+func (w *httpResponseWriterRPCServer) HjConnSetReadDeadline(t time.Time, reply *struct{}) error {
+	if w.hjr == nil {
+		return ErrNotHijacked
+	}
+	return w.hjr.conn.SetReadDeadline(t)
+}
+
+func (w *httpResponseWriterRPCServer) HjConnSetWriteDeadline(t time.Time, reply *struct{}) error {
+	if w.hjr == nil {
+		return ErrNotHijacked
+	}
+	return w.hjr.conn.SetWriteDeadline(t)
+}
+
+func (w *httpResponseWriterRPCServer) HijackResponse(args struct{}, reply *struct{}) error {
+	if w.hjr != nil {
+		return ErrAlreadyHijacked
+	}
+	hj, ok := w.w.(http.Hijacker)
+	if !ok {
+		return ErrCannotHijack
+	}
+	conn, bufrw, err := hj.Hijack()
+	if err != nil {
+		return err
+	}
+
+	w.hjr = &hijackedResponse{
+		conn:    conn,
+		bufrw:   bufrw,
+		readBuf: make([]byte, hijackedConnReadBufSize),
+	}
+	return nil
+}
+
+type hijackedConn struct {
+	client *rpc.Client
+}
+
+type hijackedConnRW struct {
+	client *rpc.Client
+}
+
+func (w *hijackedConnRW) Read(b []byte) (int, error) {
+	var data []byte
+	if err := w.client.Call("Plugin.HjConnRWRead", b, &data); err != nil {
+		return 0, err
+	}
+	copy(b, data)
+	return len(data), nil
+}
+
+func (w *hijackedConnRW) Write(b []byte) (int, error) {
+	var n int
+	if err := w.client.Call("Plugin.HjConnRWWrite", b, &n); err != nil {
+		return 0, err
+	}
+	return n, nil
+}
+
+func (w *hijackedConn) Read(b []byte) (int, error) {
+	var data []byte
+	if err := w.client.Call("Plugin.HjConnRead", len(b), &data); err != nil {
+		return 0, err
+	}
+	copy(b, data)
+	return len(data), nil
+}
+
+func (w *hijackedConn) Write(b []byte) (int, error) {
+	var n int
+	if err := w.client.Call("Plugin.HjConnWrite", b, &n); err != nil {
+		return 0, err
+	}
+	return n, nil
+}
+
+func (w *hijackedConn) Close() error {
+	return w.client.Call("Plugin.HjConnClose", struct{}{}, nil)
+}
+
+func (w *hijackedConn) LocalAddr() net.Addr {
+	return nil
+}
+
+func (w *hijackedConn) RemoteAddr() net.Addr {
+	return nil
+}
+
+func (w *hijackedConn) SetDeadline(t time.Time) error {
+	return w.client.Call("Plugin.HjConnSetDeadline", t, nil)
+}
+
+func (w *hijackedConn) SetReadDeadline(t time.Time) error {
+	return w.client.Call("Plugin.HjConnSetReadDeadline", t, nil)
+}
+
+func (w *hijackedConn) SetWriteDeadline(t time.Time) error {
+	return w.client.Call("Plugin.HjConnSetWriteDeadline", t, nil)
+}
+
+func (w *httpResponseWriterRPCClient) Hijack() (net.Conn, *bufio.ReadWriter, error) {
+	c := &hijackedConn{
+		client: w.client,
+	}
+	rw := &hijackedConnRW{
+		client: w.client,
+	}
+
+	if err := w.client.Call("Plugin.HijackResponse", struct{}{}, nil); err != nil {
+		return nil, nil, err
+	}
+
+	return c, bufio.NewReadWriter(bufio.NewReader(rw), bufio.NewWriter(rw)), nil
+}

plugin/http.go

@@ -4,18 +4,27 @@
 package plugin
 
 import (
+	"bufio"
 	"errors"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"net/rpc"
 
 	"github.com/mattermost/mattermost-server/v5/mlog"
 )
 
+type hijackedResponse struct {
+	conn    net.Conn
+	bufrw   *bufio.ReadWriter
+	readBuf []byte
+}
+
 type httpResponseWriterRPCServer struct {
 	w   http.ResponseWriter
 	log *mlog.Logger
+	hjr *hijackedResponse
 }
 
 func (w *httpResponseWriterRPCServer) Header(args struct{}, reply *http.Header) error {