neko/server/internal/websocket/manager.go

432 lines
11 KiB
Go
Raw Normal View History

2020-11-02 04:09:48 +13:00
package websocket
import (
2021-08-30 09:00:51 +12:00
"encoding/json"
"errors"
2020-11-02 04:09:48 +13:00
"net/http"
"sync"
2020-11-02 04:09:48 +13:00
"time"
"github.com/gorilla/websocket"
"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
"github.com/demodesk/neko/internal/websocket/handler"
"github.com/demodesk/neko/pkg/types"
"github.com/demodesk/neko/pkg/types/event"
"github.com/demodesk/neko/pkg/types/message"
"github.com/demodesk/neko/pkg/utils"
2020-11-02 04:09:48 +13:00
)
2021-09-03 07:37:24 +12:00
// send pings to peer with this period - must be less than pongWait
const pingPeriod = 10 * time.Second
2021-11-02 05:49:02 +13:00
// period for sending inactive cursor messages
2021-11-13 04:27:05 +13:00
const inactiveCursorsPeriod = 750 * time.Millisecond
2021-11-02 05:49:02 +13:00
// maximum payload length for logging
const maxPayloadLogLength = 10_000
2022-11-12 05:58:54 +13:00
// events that are not logged in debug mode
var nologEvents = []string{
// don't log twice
event.SYSTEM_LOGS,
// don't log heartbeat
event.SYSTEM_HEARTBEAT,
// don't log every cursor update
event.SESSION_CURSORS,
}
2020-11-02 04:09:48 +13:00
func New(
sessions types.SessionManager,
desktop types.DesktopManager,
capture types.CaptureManager,
webrtc types.WebRTCManager,
) *WebSocketManagerCtx {
logger := log.With().Str("module", "websocket").Logger()
return &WebSocketManagerCtx{
2021-02-15 02:40:17 +13:00
logger: logger,
2021-10-05 01:40:20 +13:00
shutdown: make(chan struct{}),
2021-02-15 02:40:17 +13:00
sessions: sessions,
desktop: desktop,
handler: handler.New(sessions, desktop, capture, webrtc),
2021-08-30 09:01:51 +12:00
handlers: []types.WebSocketHandler{},
2020-11-02 04:09:48 +13:00
}
}
type WebSocketManagerCtx struct {
2021-02-15 02:40:17 +13:00
logger zerolog.Logger
wg sync.WaitGroup
2021-10-05 01:40:20 +13:00
shutdown chan struct{}
2021-02-15 02:40:17 +13:00
sessions types.SessionManager
desktop types.DesktopManager
handler *handler.MessageHandlerCtx
2021-08-30 09:01:51 +12:00
handlers []types.WebSocketHandler
2022-03-27 12:08:06 +13:00
shutdownInactiveCursors chan struct{}
2020-11-02 04:09:48 +13:00
}
2021-02-15 05:11:21 +13:00
func (manager *WebSocketManagerCtx) Start() {
manager.sessions.OnCreated(func(session types.Session) {
2021-09-02 09:30:09 +12:00
err := manager.handler.SessionCreated(session)
manager.logger.Err(err).
Str("session_id", session.ID()).
Msg("session created")
2020-12-03 06:59:54 +13:00
})
2021-02-15 05:11:21 +13:00
manager.sessions.OnDeleted(func(session types.Session) {
2021-09-02 09:30:09 +12:00
err := manager.handler.SessionDeleted(session)
manager.logger.Err(err).
Str("session_id", session.ID()).
Msg("session deleted")
2020-12-03 06:59:54 +13:00
})
2021-02-15 05:11:21 +13:00
manager.sessions.OnConnected(func(session types.Session) {
2021-09-02 09:30:09 +12:00
err := manager.handler.SessionConnected(session)
manager.logger.Err(err).
Str("session_id", session.ID()).
Msg("session connected")
2020-11-02 04:09:48 +13:00
})
2021-02-15 05:11:21 +13:00
manager.sessions.OnDisconnected(func(session types.Session) {
2021-09-02 09:30:09 +12:00
err := manager.handler.SessionDisconnected(session)
manager.logger.Err(err).
Str("session_id", session.ID()).
Msg("session disconnected")
2020-11-02 04:09:48 +13:00
})
manager.sessions.OnProfileChanged(func(session types.Session, new, old types.MemberProfile) {
err := manager.handler.SessionProfileChanged(session, new, old)
2021-09-02 09:30:09 +12:00
manager.logger.Err(err).
Str("session_id", session.ID()).
Interface("new", new).
Interface("old", old).
2021-09-02 09:30:09 +12:00
Msg("session profile changed")
2020-12-03 06:59:54 +13:00
})
2021-02-15 05:11:21 +13:00
manager.sessions.OnStateChanged(func(session types.Session) {
2021-09-02 09:30:09 +12:00
err := manager.handler.SessionStateChanged(session)
manager.logger.Err(err).
Str("session_id", session.ID()).
Msg("session state changed")
2020-12-03 06:59:54 +13:00
})
2024-05-07 09:47:13 +12:00
manager.sessions.OnHostChanged(func(session, host types.Session) {
2021-09-02 07:58:39 +12:00
payload := message.ControlHost{
2024-05-07 09:47:13 +12:00
ID: session.ID(),
HasHost: host != nil,
2021-03-14 09:37:00 +13:00
}
2021-09-02 07:58:39 +12:00
if payload.HasHost {
2024-05-07 09:47:13 +12:00
payload.HostID = host.ID()
2021-03-14 09:37:00 +13:00
}
manager.sessions.Broadcast(event.CONTROL_HOST, payload)
2021-08-30 04:23:58 +12:00
2022-03-07 23:05:02 +13:00
manager.logger.Info().
2024-05-07 09:47:13 +12:00
Str("session_id", session.ID()).
2021-09-02 07:58:39 +12:00
Bool("has_host", payload.HasHost).
Str("host_id", payload.HostID).
2021-08-30 04:23:58 +12:00
Msg("session host changed")
2021-03-14 09:37:00 +13:00
})
manager.sessions.OnSettingsChanged(func(session types.Session, new, old types.Settings) {
2022-03-27 12:08:06 +13:00
// start inactive cursors
if new.InactiveCursors && !old.InactiveCursors {
manager.startInactiveCursors()
}
// stop inactive cursors
if !new.InactiveCursors && old.InactiveCursors {
manager.stopInactiveCursors()
}
2024-05-05 10:03:32 +12:00
manager.sessions.Broadcast(event.SYSTEM_SETTINGS, message.SystemSettingsUpdate{
ID: session.ID(),
Settings: new,
})
2022-03-27 12:08:06 +13:00
manager.logger.Info().
2024-05-05 10:03:32 +12:00
Str("session_id", session.ID()).
2022-03-27 12:08:06 +13:00
Interface("new", new).
Interface("old", old).
Msg("settings changed")
})
2021-02-15 05:11:21 +13:00
manager.desktop.OnClipboardUpdated(func() {
host, hasHost := manager.sessions.GetHost()
if !hasHost || !host.Profile().CanAccessClipboard {
2021-01-12 03:30:53 +13:00
return
}
2022-03-07 23:05:02 +13:00
manager.logger.Info().Msg("sync clipboard")
2021-09-02 07:58:39 +12:00
2021-02-15 05:11:21 +13:00
data, err := manager.desktop.ClipboardGetText()
2021-01-29 09:12:35 +13:00
if err != nil {
2021-09-02 09:10:06 +12:00
manager.logger.Err(err).Msg("could not get clipboard content")
2021-02-02 00:23:37 +13:00
return
2021-01-29 09:12:35 +13:00
}
host.Send(
2021-09-02 07:58:39 +12:00
event.CLIPBOARD_UPDATED,
message.ClipboardData{
Text: data.Text,
// TODO: Send HTML?
})
2021-01-12 03:30:53 +13:00
})
2021-01-12 12:09:43 +13:00
if manager.desktop.IsFileChooserDialogEnabled() {
manager.fileChooserDialogEvents()
}
2021-08-30 04:23:58 +12:00
2022-03-27 12:08:06 +13:00
if manager.sessions.Settings().InactiveCursors {
manager.startInactiveCursors()
2021-11-02 05:49:02 +13:00
}
2021-10-24 12:09:41 +13:00
2021-08-30 04:23:58 +12:00
manager.logger.Info().Msg("websocket starting")
2020-11-02 04:09:48 +13:00
}
2021-02-15 05:11:21 +13:00
func (manager *WebSocketManagerCtx) Shutdown() error {
2021-09-02 10:00:29 +12:00
manager.logger.Info().Msg("shutdown")
close(manager.shutdown)
2022-03-27 12:08:06 +13:00
manager.stopInactiveCursors()
manager.wg.Wait()
2020-11-02 04:09:48 +13:00
return nil
}
2021-08-30 09:01:51 +12:00
func (manager *WebSocketManagerCtx) AddHandler(handler types.WebSocketHandler) {
2021-02-15 05:11:21 +13:00
manager.handlers = append(manager.handlers, handler)
2020-12-13 05:40:10 +13:00
}
2021-09-19 00:59:15 +12:00
func (manager *WebSocketManagerCtx) Upgrade(checkOrigin types.CheckOrigin) types.RouterHandler {
return func(w http.ResponseWriter, r *http.Request) error {
upgrader := websocket.Upgrader{
CheckOrigin: checkOrigin,
// Do not return any error while handshake
Error: func(w http.ResponseWriter, r *http.Request, status int, reason error) {},
}
2020-11-02 04:09:48 +13:00
2021-09-19 00:59:15 +12:00
connection, err := upgrader.Upgrade(w, r, nil)
if err != nil {
return utils.HttpBadRequest().WithInternalErr(err)
}
2021-01-24 06:18:14 +13:00
2021-09-19 00:59:15 +12:00
// Cannot write HTTP response after connection upgrade
manager.connect(connection, r)
return nil
2020-11-02 04:09:48 +13:00
}
2021-09-19 00:59:15 +12:00
}
2020-11-02 04:09:48 +13:00
2021-09-19 00:59:15 +12:00
func (manager *WebSocketManagerCtx) connect(connection *websocket.Conn, r *http.Request) {
2021-03-14 08:42:56 +13:00
session, err := manager.sessions.Authenticate(r)
2020-11-02 04:09:48 +13:00
if err != nil {
2021-09-02 07:58:39 +12:00
manager.logger.Warn().Err(err).Msg("authentication failed")
newPeer(manager.logger, connection).Destroy(err.Error())
2021-09-19 00:59:15 +12:00
return
2020-12-07 06:49:08 +13:00
}
2021-09-03 06:30:50 +12:00
// add session id to all log messages
2021-09-02 07:58:39 +12:00
logger := manager.logger.With().Str("session_id", session.ID()).Logger()
// create new peer
peer := newPeer(logger, connection)
2021-08-30 04:23:58 +12:00
2021-03-14 12:45:51 +13:00
if !session.Profile().CanConnect {
2021-09-02 07:58:39 +12:00
logger.Warn().Msg("connection disabled")
2021-09-03 07:52:23 +12:00
peer.Destroy("connection disabled")
2021-09-19 00:59:15 +12:00
return
2020-11-02 04:09:48 +13:00
}
2021-03-14 12:50:08 +13:00
if session.State().IsConnected {
2021-08-30 04:23:58 +12:00
logger.Warn().Msg("already connected")
2021-08-24 04:41:19 +12:00
2022-03-27 12:08:06 +13:00
if !manager.sessions.Settings().MercifulReconnect {
2021-09-03 07:52:23 +12:00
peer.Destroy("already connected")
2021-09-19 00:59:15 +12:00
return
2021-03-16 22:45:22 +13:00
}
2021-08-30 04:23:58 +12:00
logger.Info().Msg("replacing peer connection")
2021-03-24 03:38:37 +13:00
}
2021-08-30 04:23:58 +12:00
logger.Info().
2020-11-02 04:09:48 +13:00
Str("address", connection.RemoteAddr().String()).
2021-08-30 09:03:04 +12:00
Str("agent", r.UserAgent()).
2020-11-15 04:03:39 +13:00
Msg("connection started")
2020-11-02 04:09:48 +13:00
session.ConnectWebSocketPeer(peer)
2020-11-26 08:26:11 +13:00
// this is a blocking function that lives
// throughout whole websocket connection
err = manager.handle(connection, peer, session)
2020-11-26 08:26:11 +13:00
logger.Info().
Str("address", connection.RemoteAddr().String()).
Str("agent", r.UserAgent()).
Msg("connection ended")
if err == nil {
logger.Debug().Msg("websocket close")
session.DisconnectWebSocketPeer(peer, false)
return
}
delayedDisconnect := false
e, ok := err.(*websocket.CloseError)
if !ok {
err = errors.Unwrap(err) // unwrap if possible
logger.Warn().Err(err).Msg("read message error")
// client is expected to reconnect soon
delayedDisconnect = true
} else {
switch e.Code {
case websocket.CloseNormalClosure:
logger.Debug().Str("reason", e.Text).Msg("websocket close")
case websocket.CloseGoingAway:
logger.Debug().Str("reason", "going away").Msg("websocket close")
default:
logger.Warn().Err(err).Msg("websocket close")
// abnormal websocket closure:
// client is expected to reconnect soon
delayedDisconnect = true
}
}
2020-11-02 04:09:48 +13:00
session.DisconnectWebSocketPeer(peer, delayedDisconnect)
2020-11-02 04:09:48 +13:00
}
func (manager *WebSocketManagerCtx) handle(connection *websocket.Conn, peer types.WebSocketPeer, session types.Session) error {
2021-08-30 04:23:58 +12:00
// add session id to logger context
logger := manager.logger.With().Str("session_id", session.ID()).Logger()
2020-11-02 04:09:48 +13:00
bytes := make(chan []byte)
cancel := make(chan error)
2020-11-02 04:09:48 +13:00
2021-01-24 06:18:14 +13:00
ticker := time.NewTicker(pingPeriod)
defer ticker.Stop()
2020-11-02 04:09:48 +13:00
manager.wg.Add(1)
2020-11-26 08:26:11 +13:00
go func() {
defer manager.wg.Done()
2020-11-02 04:09:48 +13:00
for {
_, raw, err := connection.ReadMessage()
2020-11-02 04:37:00 +13:00
if err != nil {
cancel <- err
2020-11-02 04:37:00 +13:00
break
2020-11-02 04:09:48 +13:00
}
2020-11-02 04:37:00 +13:00
bytes <- raw
2020-11-02 04:09:48 +13:00
}
}()
for {
select {
case raw := <-bytes:
2021-08-30 09:00:51 +12:00
data := types.WebSocketMessage{}
if err := json.Unmarshal(raw, &data); err != nil {
2021-09-02 09:10:06 +12:00
logger.Err(err).Msg("message unmarshalling has failed")
2021-08-30 09:00:51 +12:00
break
}
2022-11-12 05:58:54 +13:00
// log events if not ignored
if ok, _ := utils.ArrayIn(data.Event, nologEvents); !ok {
payload := data.Payload
if len(payload) > maxPayloadLogLength {
payload = []byte("<truncated>")
}
2021-09-10 08:51:24 +12:00
logger.Debug().
Str("address", connection.RemoteAddr().String()).
Str("event", data.Event).
Str("payload", string(payload)).
2021-09-10 08:51:24 +12:00
Msg("received message from client")
}
2020-11-02 04:09:48 +13:00
2021-08-30 09:00:51 +12:00
handled := manager.handler.Message(session, data)
2021-02-15 05:11:21 +13:00
for _, handler := range manager.handlers {
2020-12-13 05:40:10 +13:00
if handled {
break
}
2021-08-30 09:00:51 +12:00
handled = handler(session, data)
2020-12-13 05:40:10 +13:00
}
if !handled {
2021-08-30 09:00:51 +12:00
logger.Warn().Str("event", data.Event).Msg("unhandled message")
2020-11-02 04:09:48 +13:00
}
case err := <-cancel:
return err
case <-manager.shutdown:
2021-09-25 01:15:59 +12:00
peer.Destroy("connection shutdown")
return nil
2020-11-02 04:09:48 +13:00
case <-ticker.C:
2021-09-25 01:15:59 +12:00
if err := peer.Ping(); err != nil {
return err
2020-11-02 04:09:48 +13:00
}
}
}
}
2021-11-04 11:52:44 +13:00
2022-03-27 12:08:06 +13:00
func (manager *WebSocketManagerCtx) startInactiveCursors() {
if manager.shutdownInactiveCursors != nil {
manager.logger.Warn().Msg("inactive cursors handler already running")
return
}
2021-11-04 11:52:44 +13:00
manager.logger.Info().Msg("starting inactive cursors handler")
2022-03-27 12:08:06 +13:00
manager.shutdownInactiveCursors = make(chan struct{})
2021-11-04 11:52:44 +13:00
manager.wg.Add(1)
go func() {
defer manager.wg.Done()
ticker := time.NewTicker(inactiveCursorsPeriod)
defer ticker.Stop()
2021-11-08 09:23:23 +13:00
var currentEmpty bool
var lastEmpty = false
2021-11-04 11:52:44 +13:00
for {
select {
2022-03-27 12:08:06 +13:00
case <-manager.shutdownInactiveCursors:
2021-11-04 11:52:44 +13:00
manager.logger.Info().Msg("stopping inactive cursors handler")
2022-03-27 12:08:06 +13:00
manager.shutdownInactiveCursors = nil
2022-03-27 12:26:11 +13:00
// remove last cursor entries and send empty message
_ = manager.sessions.PopCursors()
manager.sessions.InactiveCursorsBroadcast(event.SESSION_CURSORS, []message.SessionCursors{})
2021-11-04 11:52:44 +13:00
return
case <-ticker.C:
cursorsMap := manager.sessions.PopCursors()
2021-11-08 09:23:23 +13:00
currentEmpty = len(cursorsMap) == 0
if currentEmpty && lastEmpty {
2021-11-04 11:52:44 +13:00
continue
}
2021-11-08 09:23:23 +13:00
lastEmpty = currentEmpty
2021-11-04 11:52:44 +13:00
2021-11-13 04:27:05 +13:00
sessionCursors := []message.SessionCursors{}
for session, cursors := range cursorsMap {
sessionCursors = append(
sessionCursors,
message.SessionCursors{
ID: session.ID(),
Cursors: cursors,
2021-11-04 11:52:44 +13:00
},
)
}
manager.sessions.InactiveCursorsBroadcast(event.SESSION_CURSORS, sessionCursors)
2021-11-04 11:52:44 +13:00
}
}
}()
}
2022-03-27 12:08:06 +13:00
func (manager *WebSocketManagerCtx) stopInactiveCursors() {
if manager.shutdownInactiveCursors != nil {
close(manager.shutdownInactiveCursors)
}
}