self-hosted/focalboard
1
0
Fork 0
mirror of https://github.com/mattermost/focalboard.git synced 2024-12-24 13:43:12 +02:00
Code Issues Releases Activity

Websocket auth

Browse Source
This commit is contained in:
Chen-I Lim 2021-02-02 12:06:28 -08:00
parent 12846ccc38
commit b3e660d354
5 changed files with 211 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
server/app/blocks_test.go
View File

@ -5,6 +5,7 @@ import (
"testing"
"github.com/golang/mock/gomock"
"github.com/mattermost/focalboard/server/auth"
"github.com/mattermost/focalboard/server/services/config"
"github.com/mattermost/focalboard/server/services/store/mockstore"
"github.com/mattermost/focalboard/server/services/webhook"
@ -13,12 +14,17 @@ import (
"github.com/stretchr/testify/require"
)
func isValidSessionToken(token string) bool {
return true
}
func TestGetParentID(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
cfg := config.Configuration{}
store := mockstore.NewMockStore(ctrl)
wsserver := ws.NewServer()
auth := auth.New(&cfg, store)
wsserver := ws.NewServer(auth, true)
webhook := webhook.NewClient(&cfg)
app := New(&cfg, store, wsserver, &mocks.FileBackend{}, webhook)

37
server/auth/auth.go Normal file
View File

@ -0,0 +1,37 @@
package auth
import (
"time"
"github.com/mattermost/focalboard/server/model"
"github.com/mattermost/focalboard/server/services/config"
"github.com/mattermost/focalboard/server/services/store"
"github.com/pkg/errors"
)
// Auth authenticates sessions
type Auth struct {
config *config.Configuration
store store.Store
}
// New returns a new Auth
func New(config *config.Configuration, store store.Store) *Auth {
return &Auth{config: config, store: store}
}
// GetSession Get a user active session and refresh the session if is needed
func (a *Auth) GetSession(token string) (*model.Session, error) {
if len(token) < 1 {
return nil, errors.New("no session token")
}
session, err := a.store.GetSession(token, a.config.SessionExpireTime)
if err != nil {
return nil, errors.Wrap(err, "unable to get the session for the token")
}
if session.UpdateAt < (time.Now().Unix() - a.config.SessionRefreshTime) {
a.store.RefreshSession(session)
}
return session, nil
}

5
server/server/server.go
View File

@ -17,6 +17,7 @@ import (
"github.com/mattermost/focalboard/server/api"
"github.com/mattermost/focalboard/server/app"
"github.com/mattermost/focalboard/server/auth"
"github.com/mattermost/focalboard/server/context"
appModel "github.com/mattermost/focalboard/server/model"
"github.com/mattermost/focalboard/server/services/config"
@ -60,7 +61,9 @@ func New(cfg *config.Configuration, singleUser bool) (*Server, error) {
return nil, err
}
wsServer := ws.NewServer()
auth := auth.New(cfg, store)
wsServer := ws.NewServer(auth, singleUser)
filesBackendSettings := model.FileSettings{}
filesBackendSettings.SetDefaults(false)

232
server/ws/websockets.go
View File

@ -9,95 +9,20 @@ import (
"github.com/gorilla/mux"
"github.com/gorilla/websocket"
"github.com/mattermost/focalboard/server/auth"
"github.com/mattermost/focalboard/server/model"
)
// RegisterRoutes registers routes.
func (ws *Server) RegisterRoutes(r *mux.Router) {
r.HandleFunc("/ws/onchange", ws.handleWebSocketOnChange)
}
// AddListener adds a listener for a block's change.
func (ws *Server) AddListener(client *websocket.Conn, blockIDs []string) {
ws.mu.Lock()
for _, blockID := range blockIDs {
if ws.listeners[blockID] == nil {
ws.listeners[blockID] = []*websocket.Conn{}
}
ws.listeners[blockID] = append(ws.listeners[blockID], client)
}
ws.mu.Unlock()
}
// RemoveListener removes a webSocket listener from all blocks.
func (ws *Server) RemoveListener(client *websocket.Conn) {
ws.mu.Lock()
for key, clients := range ws.listeners {
listeners := []*websocket.Conn{}
for _, existingClient := range clients {
if client != existingClient {
listeners = append(listeners, existingClient)
}
}
ws.listeners[key] = listeners
}
ws.mu.Unlock()
}
// RemoveListenerFromBlocks removes a webSocket listener from a set of block.
func (ws *Server) RemoveListenerFromBlocks(client *websocket.Conn, blockIDs []string) {
ws.mu.Lock()
for _, blockID := range blockIDs {
listeners := ws.listeners[blockID]
if listeners == nil {
return
}
// Remove the first instance of this client that's listening to this block
// Note: A client can listen multiple times to the same block
for index, listener := range listeners {
if client == listener {
newListeners := append(listeners[:index], listeners[index+1:]...)
ws.listeners[blockID] = newListeners
break
}
}
}
ws.mu.Unlock()
}
// GetListeners returns the listeners to a blockID's changes.
func (ws *Server) GetListeners(blockID string) []*websocket.Conn {
ws.mu.Lock()
listeners := ws.listeners[blockID]
ws.mu.Unlock()
return listeners
}
// IsValidSessionToken authenticates session tokens
type IsValidSessionToken func(token string) bool
// Server is a WebSocket server.
type Server struct {
upgrader websocket.Upgrader
listeners map[string][]*websocket.Conn
mu sync.RWMutex
}
// NewServer creates a new Server.
func NewServer() *Server {
return &Server{
listeners: make(map[string][]*websocket.Conn),
upgrader: websocket.Upgrader{
CheckOrigin: func(r *http.Request) bool {
return true
},
},
}
upgrader websocket.Upgrader
listeners map[string][]*websocket.Conn
mu sync.RWMutex
auth *auth.Auth
singleUser bool
}
// UpdateMsg is sent on block updates
@ -109,9 +34,34 @@ type UpdateMsg struct {
// WebsocketCommand is an incoming command from the client.
type WebsocketCommand struct {
Action string `json:"action"`
Token string `json:"token"`
BlockIDs []string `json:"blockIds"`
}
type websocketSession struct {
client *websocket.Conn
isAuthenticated bool
}
// NewServer creates a new Server.
func NewServer(auth *auth.Auth, singleUser bool) *Server {
return &Server{
listeners: make(map[string][]*websocket.Conn),
upgrader: websocket.Upgrader{
CheckOrigin: func(r *http.Request) bool {
return true
},
},
auth: auth,
singleUser: singleUser,
}
}
// RegisterRoutes registers routes.
func (ws *Server) RegisterRoutes(r *mux.Router) {
r.HandleFunc("/ws/onchange", ws.handleWebSocketOnChange)
}
func (ws *Server) handleWebSocketOnChange(w http.ResponseWriter, r *http.Request) {
// Upgrade initial GET request to a websocket
client, err := ws.upgrader.Upgrade(w, r, nil)
@ -128,17 +78,22 @@ func (ws *Server) handleWebSocketOnChange(w http.ResponseWriter, r *http.Request
log.Printf("DISCONNECT WebSocket onChange, client: %s", client.RemoteAddr())
// Remove client from listeners
ws.RemoveListener(client)
ws.removeListener(client)
client.Close()
}()
wsSession := websocketSession{
client: client,
isAuthenticated: ws.singleUser,
}
// Simple message handling loop
for {
_, p, err := client.ReadMessage()
if err != nil {
log.Printf("ERROR WebSocket onChange, client: %s, err: %v", client.RemoteAddr(), err)
ws.RemoveListener(client)
ws.removeListener(client)
break
}
@ -154,13 +109,17 @@ func (ws *Server) handleWebSocketOnChange(w http.ResponseWriter, r *http.Request
}
switch command.Action {
case "AUTH":
log.Printf(`Command: AUTH, client: %s`, client.RemoteAddr())
ws.authenticateListener(&wsSession, command.Token)
case "ADD":
log.Printf(`Command: Add blockID: %v, client: %s`, command.BlockIDs, client.RemoteAddr())
ws.AddListener(client, command.BlockIDs)
ws.addListener(&wsSession, command.BlockIDs)
case "REMOVE":
log.Printf(`Command: Remove blockID: %v, client: %s`, command.BlockIDs, client.RemoteAddr())
ws.RemoveListenerFromBlocks(client, command.BlockIDs)
ws.removeListenerFromBlocks(&wsSession, command.BlockIDs)
default:
log.Printf(`ERROR webSocket command, invalid action: %v`, command.Action)
@ -168,6 +127,105 @@ func (ws *Server) handleWebSocketOnChange(w http.ResponseWriter, r *http.Request
}
}
func (ws *Server) isValidSessionToken(token string) bool {
if ws.singleUser {
return true
}
session, err := ws.auth.GetSession(token)
if session == nil || err != nil {
return false
}
return true
}
func (ws *Server) authenticateListener(wsSession *websocketSession, token string) {
isValidSession := ws.isValidSessionToken(token)
if !isValidSession {
wsSession.client.Close()
return
}
// Authenticated
wsSession.isAuthenticated = true
log.Printf("authenticateListener: Authenticated")
}
// AddListener adds a listener for a block's change.
func (ws *Server) addListener(wsSession *websocketSession, blockIDs []string) {
if !wsSession.isAuthenticated {
log.Printf("addListener: NOT AUTHENTICATED")
return
}
ws.mu.Lock()
for _, blockID := range blockIDs {
if ws.listeners[blockID] == nil {
ws.listeners[blockID] = []*websocket.Conn{}
}
ws.listeners[blockID] = append(ws.listeners[blockID], wsSession.client)
}
ws.mu.Unlock()
}
// removeListener removes a webSocket listener from all blocks.
func (ws *Server) removeListener(client *websocket.Conn) {
ws.mu.Lock()
for key, clients := range ws.listeners {
listeners := []*websocket.Conn{}
for _, existingClient := range clients {
if client != existingClient {
listeners = append(listeners, existingClient)
}
}
ws.listeners[key] = listeners
}
ws.mu.Unlock()
}
// removeListenerFromBlocks removes a webSocket listener from a set of block.
func (ws *Server) removeListenerFromBlocks(wsSession *websocketSession, blockIDs []string) {
if !wsSession.isAuthenticated {
log.Printf("removeListenerFromBlocks: NOT AUTHENTICATED")
return
}
ws.mu.Lock()
for _, blockID := range blockIDs {
listeners := ws.listeners[blockID]
if listeners == nil {
return
}
// Remove the first instance of this client that's listening to this block
// Note: A client can listen multiple times to the same block
for index, listener := range listeners {
if wsSession.client == listener {
newListeners := append(listeners[:index], listeners[index+1:]...)
ws.listeners[blockID] = newListeners
break
}
}
}
ws.mu.Unlock()
}
// GetListeners returns the listeners to a blockID's changes.
func (ws *Server) GetListeners(blockID string) []*websocket.Conn {
ws.mu.Lock()
listeners := ws.listeners[blockID]
ws.mu.Unlock()
return listeners
}
// BroadcastBlockDelete broadcasts delete messages to clients
func (ws *Server) BroadcastBlockDelete(blockID string, parentID string) {
now := time.Now().Unix()

24
webapp/src/octoListener.ts
View File

@ -27,6 +27,7 @@ class OctoListener {
}
readonly serverUrl: string
private token: string
private ws?: WebSocket
private blockIds: string[] = []
private isInitialized = false
@ -38,14 +39,13 @@ class OctoListener {
notificationDelay = 100
reopenDelay = 3000
constructor(serverUrl?: string) {
constructor(serverUrl?: string, token?: string) {
this.serverUrl = serverUrl || window.location.origin
this.token = token || localStorage.getItem('sessionId') || ''
Utils.log(`OctoListener serverUrl: ${this.serverUrl}`)
}
open(blockIds: string[], onChange: OnChangeHandler, onReconnect: () => void): void {
let timeoutId: NodeJS.Timeout
if (this.ws) {
this.close()
}
@ -61,6 +61,7 @@ class OctoListener {
ws.onopen = () => {
Utils.log('OctoListener webSocket opened.')
this.authenticate()
this.addBlocks(blockIds)
this.isInitialized = true
}
@ -93,9 +94,6 @@ class OctoListener {
const message = JSON.parse(e.data) as WSMessage
switch (message.action) {
case 'UPDATE_BLOCK':
if (timeoutId) {
clearTimeout(timeoutId)
}
Utils.log(`OctoListener update block: ${message.block?.id}`)
this.queueUpdateNotification(message.block)
break
@ -124,6 +122,20 @@ class OctoListener {
ws.close()
}
authenticate(): void {
if (!this.ws) {
Utils.assertFailure('OctoListener.addBlocks: ws is not open')
return
}
const command = {
action: 'AUTH',
token: this.token,
}
this.ws.send(JSON.stringify(command))
}
addBlocks(blockIds: string[]): void {
if (!this.ws) {
Utils.assertFailure('OctoListener.addBlocks: ws is not open')