revertws (#17216)

* Revert "MM-34000: Use non-epoll mode for TLS connections (#17172)" This reverts commit 2743089b54. * Revert "MM-33233: Fix double close of webconn pump (#17026)" This reverts commit 0f98620b65. * Revert "MM-33836: Detect and upgrade incorrect HTTP version for websocket handshakes (#17142)" This reverts commit 4c5ea07aff. * revert i18n * Revert "MM-21012: Revamp websocket implementation (#16620)" This reverts commit a246104d04. * fix go.mod * Trigger CI
2021-03-24 15:29:23 +05:30
--- a/app/app_iface.go
+++ b/app/app_iface.go
@@ -12,12 +12,12 @@ import (
 	"crypto/ecdsa"
 	"io"
 	"mime/multipart"
-	"net"
 	"net/http"
 	"net/url"
 	"time"

 	"github.com/dyatlov/go-opengraph/opengraph"
+	"github.com/gorilla/websocket"
 	"github.com/mattermost/mattermost-server/v5/audit"
 	"github.com/mattermost/mattermost-server/v5/einterfaces"
 	"github.com/mattermost/mattermost-server/v5/model"
@@ -234,7 +234,7 @@ type AppIface interface {
 	// function is only exposed to sysadmins and the possibility of this edge case is relatively small.
 	MoveChannel(team *model.Team, channel *model.Channel, user *model.User) *model.AppError
 	// NewWebConn returns a new WebConn instance.
-	NewWebConn(ws net.Conn, session model.Session, t i18n.TranslateFunc, locale string) *WebConn
+	NewWebConn(ws *websocket.Conn, session model.Session, t i18n.TranslateFunc, locale string) *WebConn
 	// NewWebHub creates a new Hub.
 	NewWebHub() *Hub
 	// NotifySessionsExpired is called periodically from the job server to notify any mobile sessions that have expired.
--- a/app/opentracing/opentracing_layer.go
+++ b/app/opentracing/opentracing_layer.go
@@ -12,12 +12,12 @@ import (
 	"crypto/ecdsa"
 	"io"
 	"mime/multipart"
-	"net"
 	"net/http"
 	"net/url"
 	"time"

 	"github.com/dyatlov/go-opengraph/opengraph"
+	"github.com/gorilla/websocket"
 	"github.com/mattermost/mattermost-server/v5/app"
 	"github.com/mattermost/mattermost-server/v5/audit"
 	"github.com/mattermost/mattermost-server/v5/einterfaces"
@@ -10874,7 +10874,7 @@ func (a *OpenTracingAppLayer) NewPluginAPI(manifest *model.Manifest) plugin.API
 	return resultVar0
 }

-func (a *OpenTracingAppLayer) NewWebConn(ws net.Conn, session model.Session, t i18n.TranslateFunc, locale string) *app.WebConn {
+func (a *OpenTracingAppLayer) NewWebConn(ws *websocket.Conn, session model.Session, t i18n.TranslateFunc, locale string) *app.WebConn {
 	origCtx := a.ctx
 	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.NewWebConn")

--- a/app/server.go
+++ b/app/server.go
@@ -33,7 +33,6 @@ import (
 	sentryhttp "github.com/getsentry/sentry-go/http"
 	"github.com/gorilla/handlers"
 	"github.com/gorilla/mux"
-	"github.com/mailru/easygo/netpoll"
 	"github.com/pkg/errors"
 	"github.com/rs/cors"
 	"golang.org/x/crypto/acme/autocert"
@@ -113,11 +112,8 @@ type Server struct {

 	EmailService *EmailService

-	hubs          []*Hub
-	hashSeed      maphash.Seed
-	poller        netpoll.Poller
-	webConnSema   chan struct{}
-	webConnSemaWg sync.WaitGroup
+	hubs     []*Hub
+	hashSeed maphash.Seed

 	PushNotificationsHub   PushNotificationsHub
 	pushNotificationClient *http.Client // TODO: move this to it's own package
@@ -240,16 +236,6 @@ func NewServer(options ...Option) (*Server, error) {
 		mlog.Error("Could not initiate logging", mlog.Err(err))
 	}

-	// epoll/kqueue is not available on Windows.
-	if runtime.GOOS != "windows" {
-		poller, err := netpoll.New(nil)
-		if err != nil {
-			return nil, errors.Wrap(err, "failed to create a netpoll instance")
-		}
-		s.poller = poller
-		s.webConnSema = make(chan struct{}, runtime.NumCPU()*8) // numCPU * 8 is a good amount of concurrency.
-	}
-
 	// This is called after initLogging() to avoid a race condition.
 	mlog.Info("Server is initializing...", mlog.String("go_version", runtime.Version()))

@@ -283,9 +269,9 @@ func NewServer(options ...Option) (*Server, error) {
 	}

 	if *s.Config().ServiceSettings.EnableOpenTracing {
-		tracer, err2 := tracing.New()
-		if err2 != nil {
-			return nil, err2
+		tracer, err := tracing.New()
+		if err != nil {
+			return nil, err
 		}
 		s.tracer = tracer
 	}
@@ -1284,7 +1270,7 @@ func (a *App) OriginChecker() func(*http.Request) bool {

 		return utils.OriginChecker(allowed)
 	}
-	return utils.SameOriginChecker()
+	return nil
 }

 func (s *Server) checkPushNotificationServerUrl() {
@@ -1796,10 +1782,6 @@ func (s *Server) SetLog(l *mlog.Logger) {
 	s.Log = l
 }

-func (s *Server) Poller() netpoll.Poller {
-	return s.poller
-}
-
 func (a *App) GenerateSupportPacket() []model.FileData {
 	// If any errors we come across within this function, we will log it in a warning.txt file so that we know why certain files did not get produced if any
 	var warnings []string
--- a/app/web_conn.go
+++ b/app/web_conn.go
@@ -7,18 +7,12 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
-	"net"
 	"net/http"
-	"runtime"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"time"

-	"github.com/gobwas/ws"
-	"github.com/gobwas/ws/wsutil"
-	"github.com/mailru/easygo/netpoll"
-	"github.com/pkg/errors"
+	"github.com/gorilla/websocket"

 	"github.com/mattermost/mattermost-server/v5/model"
 	"github.com/mattermost/mattermost-server/v5/shared/i18n"
@@ -36,35 +30,30 @@ const (
 	webConnMemberCacheTime = 1000 * 60 * 30 // 30 minutes
 )

-var errNonEpollConnClose = errors.New("connection closed")
-
 // WebConn represents a single websocket connection to a user.
-// It contains all the necessary state to manage sending/receiving data to/from
+// It contains all the necesarry state to manage sending/receiving data to/from
 // a websocket.
 type WebConn struct {
 	sessionExpiresAt int64 // This should stay at the top for 64-bit alignment of 64-bit words accessed atomically
 	App              *App
-	WebSocket        net.Conn
+	WebSocket        *websocket.Conn
 	T                i18n.TranslateFunc
 	Locale           string
 	Sequence         int64
 	UserId           string

-	readMut                   sync.Mutex
 	allChannelMembers         map[string]string
 	lastAllChannelMembersTime int64
 	lastUserActivityAt        int64
 	send                      chan model.WebSocketMessage
 	sessionToken              atomic.Value
 	session                   atomic.Value
-	hasEpoll                  bool
 	endWritePump              chan struct{}
 	pumpFinished              chan struct{}
-	closeOnce                 sync.Once
 }

 // NewWebConn returns a new WebConn instance.
-func (a *App) NewWebConn(ws net.Conn, session model.Session, t i18n.TranslateFunc, locale string) *WebConn {
+func (a *App) NewWebConn(ws *websocket.Conn, session model.Session, t i18n.TranslateFunc, locale string) *WebConn {
 	if session.UserId != "" {
 		a.Srv().Go(func() {
 			a.SetStatusOnline(session.UserId, false)
@@ -80,7 +69,6 @@ func (a *App) NewWebConn(ws net.Conn, session model.Session, t i18n.TranslateFun
 		UserId:             session.UserId,
 		T:                  t,
 		Locale:             locale,
-		hasEpoll:           *a.Config().ServiceSettings.ConnectionSecurity == "" && runtime.GOOS != "windows",
 		endWritePump:       make(chan struct{}),
 		pumpFinished:       make(chan struct{}),
 	}
@@ -89,27 +77,13 @@ func (a *App) NewWebConn(ws net.Conn, session model.Session, t i18n.TranslateFun
 	wc.SetSessionToken(session.Token)
 	wc.SetSessionExpiresAt(session.ExpiresAt)

-	if wc.hasEpoll {
-		wc.startPoller()
-	}
 	return wc
 }

 // Close closes the WebConn.
-// It is made idempotent in nature by using a sync.Once
-// to avoid a race condition that happens when an EventReadHup event
-// and a connection close event happens at the same time.
 func (wc *WebConn) Close() {
-	wc.closeOnce.Do(func() {
-		wc.WebSocket.Close()
-		if wc.hasEpoll {
-			// This triggers the pump exit.
-			// If the pump has already exited, this just becomes a noop.
-			close(wc.endWritePump)
-		}
-		// We wait for the pump to fully exit.
-		<-wc.pumpFinished
-	})
+	wc.WebSocket.Close()
+	<-wc.pumpFinished
 }

 // GetSessionExpiresAt returns the time at which the session expires.
@@ -146,26 +120,9 @@ func (wc *WebConn) SetSession(v *model.Session) {
 	wc.session.Store(v)
 }

-// Epoll returns whether the websocket is eligible to use epoll or not.
-func (wc *WebConn) Epoll() bool {
-	return wc.hasEpoll
-}
-
 // Pump starts the WebConn instance. After this, the websocket
-// is ready to send messages.
-// This is only used by *nix platforms.
+// is ready to send/receive messages.
 func (wc *WebConn) Pump() {
-	// writePump is blocking in nature.
-	wc.writePump()
-	// Once it exits, we close everything.
-	wc.App.HubUnregister(wc)
-	close(wc.pumpFinished)
-}
-
-// BlockingPump is the non-epoll alternative of Pump.
-// It creates two goroutines - one for reading, another
-// for writing.
-func (wc *WebConn) BlockingPump() {
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go func() {
@@ -181,110 +138,29 @@ func (wc *WebConn) BlockingPump() {
 	defer ReturnSessionToPool(wc.GetSession())
 }

-// startPoller adds the file descriptor of the connection
-// to the global epoll instance and registers a callback.
-func (wc *WebConn) startPoller() {
-	desc := netpoll.Must(netpoll.HandleRead(wc.WebSocket))
-	wc.App.Srv().Poller().Start(desc, func(wsEv netpoll.Event) {
-		if wsEv&(netpoll.EventReadHup|netpoll.EventHup) != 0 {
-			wc.App.Srv().Poller().Stop(desc)
-			wc.Close()
-			return
-		}
-
-		// Block until we have a token.
-		wc.App.Srv().GetWebConnToken()
-		// Read from conn.
-		go func() {
-			defer wc.App.Srv().ReleaseWebConnToken()
-			err := wc.ReadMsg()
-			if err != nil {
-				mlog.Debug("Error while reading message from websocket", mlog.Err(err))
-				wc.App.Srv().Poller().Stop(desc)
-				// net.ErrClosed is not available until Go 1.16.
-				// https://github.com/golang/go/issues/4373
-				//
-				// Sometimes, the netpoller generates a data event and a HUP event
-				// close to each other. In that case, we don't want to double-close
-				// the connection.
-				if !strings.Contains(err.Error(), "use of closed network connection") {
-					wc.Close()
-				}
-			}
-		}()
-	})
-}
-
-// GetWebConnToken creates backpressure by using
-// a counting semaphore to limit the number of concurrent goroutines.
-func (s *Server) GetWebConnToken() {
-	s.webConnSemaWg.Add(1)
-	s.webConnSema <- struct{}{}
-}
-
-// ReleaseWebConnToken releases a token
-// got from the semaphore
-func (s *Server) ReleaseWebConnToken() {
-	<-s.webConnSema
-	s.webConnSemaWg.Done()
-}
-
-// ReadMsg will read a single message from the websocket connection.
-func (wc *WebConn) ReadMsg() error {
-	r := wsutil.NewReader(wc.WebSocket, ws.StateServerSide)
-	r.MaxFrameSize = model.SOCKET_MAX_MESSAGE_SIZE_KB
-	decoder := json.NewDecoder(r)
-
-	// The reader's methods are not goroutine safe.
-	// We restrict only one reader goroutine per-connection.
-	wc.readMut.Lock()
-	defer wc.readMut.Unlock()
-
-	var req model.WebSocketRequest
-	hdr, err := r.NextFrame()
-	if err != nil {
-		return errors.Wrap(err, "error while getting the next websocket frame")
-	}
-	switch hdr.OpCode {
-	case ws.OpClose:
-		// Return if closed.
-		// We need to return an error for non-epoll systems to let the reader exit.
-		if !wc.hasEpoll {
-			return errNonEpollConnClose
-		}
-		return nil
-	case ws.OpPong:
+func (wc *WebConn) readPump() {
+	defer func() {
+		wc.WebSocket.Close()
+	}()
+	wc.WebSocket.SetReadLimit(model.SOCKET_MAX_MESSAGE_SIZE_KB)
+	wc.WebSocket.SetReadDeadline(time.Now().Add(pongWaitTime))
+	wc.WebSocket.SetPongHandler(func(string) error {
 		wc.WebSocket.SetReadDeadline(time.Now().Add(pongWaitTime))
-		// Handle pongs
 		if wc.IsAuthenticated() {
 			wc.App.Srv().Go(func() {
 				wc.App.SetStatusAwayIfNeeded(wc.UserId, false)
 			})
 		}
-	default:
-		// Default case of data message.
-		if err := decoder.Decode(&req); err != nil {
-			// We discard any remaining data left in the socket.
-			r.Discard()
-			return errors.Wrap(err, "error during decoding websocket message")
-		}
-
-		wc.App.Srv().WebSocketRouter.ServeWebSocket(wc, &req)
-	}
-	return nil
-}
-
-func (wc *WebConn) readPump() {
-	defer wc.WebSocket.Close()
-	wc.WebSocket.SetReadDeadline(time.Now().Add(pongWaitTime))
+		return nil
+	})

 	for {
-		if err := wc.ReadMsg(); err != nil {
-			if err != errNonEpollConnClose {
-				wc.logSocketErr("websocket.read", err)
-			}
+		var req model.WebSocketRequest
+		if err := wc.WebSocket.ReadJSON(&req); err != nil {
+			wc.logSocketErr("websocket.read", err)
 			return
 		}
+		wc.App.Srv().WebSocketRouter.ServeWebSocket(wc, &req)
 	}
 }

@@ -308,7 +184,7 @@ func (wc *WebConn) writePump() {
 		case msg, ok := <-wc.send:
 			if !ok {
 				wc.WebSocket.SetWriteDeadline(time.Now().Add(writeWaitTime))
-				wsutil.WriteServerMessage(wc.WebSocket, ws.OpClose, []byte{})
+				wc.WebSocket.WriteMessage(websocket.CloseMessage, []byte{})
 				return
 			}

@@ -363,7 +239,7 @@ func (wc *WebConn) writePump() {
 			}

 			wc.WebSocket.SetWriteDeadline(time.Now().Add(writeWaitTime))
-			if err := wsutil.WriteServerMessage(wc.WebSocket, ws.OpText, buf.Bytes()); err != nil {
+			if err := wc.WebSocket.WriteMessage(websocket.TextMessage, buf.Bytes()); err != nil {
 				wc.logSocketErr("websocket.send", err)
 				return
 			}
@@ -373,7 +249,7 @@ func (wc *WebConn) writePump() {
 			}
 		case <-ticker.C:
 			wc.WebSocket.SetWriteDeadline(time.Now().Add(writeWaitTime))
-			if err := wsutil.WriteServerMessage(wc.WebSocket, ws.OpPing, []byte{}); err != nil {
+			if err := wc.WebSocket.WriteMessage(websocket.PingMessage, []byte{}); err != nil {
 				wc.logSocketErr("websocket.ticker", err)
 				return
 			}
@@ -560,5 +436,10 @@ func (wc *WebConn) isMemberOfTeam(teamID string) bool {
 }

 func (wc *WebConn) logSocketErr(source string, err error) {
-	mlog.Debug(source+": error during writing to websocket", mlog.String("user_id", wc.UserId), mlog.Err(err))
+	// browsers will appear as CloseNoStatusReceived
+	if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) {
+		mlog.Debug(source+": client side closed socket", mlog.String("user_id", wc.UserId))
+	} else {
+		mlog.Debug(source+": closing websocket", mlog.String("user_id", wc.UserId), mlog.Err(err))
+	}
 }
--- a/app/web_hub.go
+++ b/app/web_hub.go
@@ -114,9 +114,6 @@ func (a *App) InvalidateWebConnSessionCacheForUser(userID string) {
 func (s *Server) HubStop() {
 	mlog.Info("stopping websocket hub connections")

-	// Wait until all messages have finished reading.
-	s.webConnSemaWg.Wait()
-	// Now stop the hub.
 	for _, hub := range s.hubs {
 		hub.Stop()
 	}
--- a/app/web_hub_test.go
+++ b/app/web_hub_test.go
@@ -4,17 +4,13 @@
 package app

 import (
-	"context"
-	"io"
-	"io/ioutil"
 	"net"
 	"net/http"
 	"net/http/httptest"
 	"testing"
 	"time"

-	"github.com/gobwas/ws"
-	"github.com/gobwas/ws/wsutil"
+	"github.com/gorilla/websocket"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
@@ -26,35 +22,16 @@ import (

 func dummyWebsocketHandler(t *testing.T) http.HandlerFunc {
 	return func(w http.ResponseWriter, req *http.Request) {
-		upgrader := ws.HTTPUpgrader{
-			Timeout: 5 * time.Second,
-		}
-
-		var hdr ws.Header
-		conn, _, _, err := upgrader.Upgrade(req, w)
-		rd := wsutil.Reader{
-			Source:          conn,
-			State:           ws.StateServerSide,
-			CheckUTF8:       true,
-			SkipHeaderCheck: true,
+		upgrader := &websocket.Upgrader{
+			ReadBufferSize:  1024,
+			WriteBufferSize: 1024,
 		}
+		conn, err := upgrader.Upgrade(w, req, nil)
 		for err == nil {
-			hdr, err = rd.NextFrame()
-			if err != nil {
-				continue
-			}
-			if hdr.OpCode.IsControl() {
-				continue
-			}
-			if hdr.OpCode&(ws.OpText|ws.OpBinary) == 0 {
-				err = rd.Discard()
-				continue
-			}
-
-			_, err = ioutil.ReadAll(&rd)
+			_, _, err = conn.ReadMessage()
 		}
-		if err != io.EOF {
-			require.Fail(t, "unexpected error:", err)
+		if _, ok := err.(*websocket.CloseError); !ok {
+			require.NoError(t, err)
 		}
 	}
 }
@@ -65,7 +42,8 @@ func registerDummyWebConn(t *testing.T, a *App, addr net.Addr, userID string) *W
 	})
 	require.Nil(t, appErr)

-	c, _, _, err := ws.Dial(context.Background(), "ws://"+addr.String()+"/ws")
+	d := websocket.Dialer{}
+	c, _, err := d.Dial("ws://"+addr.String()+"/ws", nil)
 	require.NoError(t, err)

 	wc := a.NewWebConn(c, *session, i18n.IdentityTfunc(), "en")
@@ -90,20 +68,6 @@ func TestHubStopWithMultipleConnections(t *testing.T) {
 	defer wc3.Close()
 }

-func TestWebConnDoubleClose(t *testing.T) {
-	th := Setup(t)
-	defer th.TearDown()
-
-	s := httptest.NewServer(dummyWebsocketHandler(t))
-	defer s.Close()
-
-	wc1 := registerDummyWebConn(t, th.App, s.Listener.Addr(), "userID")
-	wc1.Close()
-	require.NotPanics(t, func() {
-		wc1.Close()
-	})
-}
-
 // TestHubStopRaceCondition verifies that attempts to use the hub after it has shutdown does not
 // block the caller indefinitely.
 func TestHubStopRaceCondition(t *testing.T) {
@@ -114,7 +78,8 @@ func TestHubStopRaceCondition(t *testing.T) {
 	s := httptest.NewServer(dummyWebsocketHandler(t))

 	th.App.HubStart()
-	registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id)
+	wc1 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id)
+	defer wc1.Close()

 	hub := th.App.Srv().hubs[0]
 	th.App.HubStop()