neko/internal/websocket/manager.go

294 lines
7.1 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"
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"
2021-01-10 11:01:39 +13:00
"demodesk/neko/internal/types"
2020-11-02 04:09:48 +13:00
"demodesk/neko/internal/types/event"
"demodesk/neko/internal/types/message"
2021-09-19 00:59:15 +12:00
"demodesk/neko/internal/utils"
2021-02-15 02:40:17 +13:00
"demodesk/neko/internal/websocket/handler"
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
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
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
})
2021-02-15 05:11:21 +13:00
manager.sessions.OnProfileChanged(func(session types.Session) {
2021-09-02 09:30:09 +12:00
err := manager.handler.SessionProfileChanged(session)
manager.logger.Err(err).
Str("session_id", session.ID()).
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
})
2021-03-14 09:37:00 +13:00
manager.sessions.OnHostChanged(func(session types.Session) {
2021-09-02 07:58:39 +12:00
payload := message.ControlHost{
2021-03-14 09:37:00 +13:00
HasHost: session != nil,
}
2021-09-02 07:58:39 +12:00
if payload.HasHost {
payload.HostID = session.ID()
2021-03-14 09:37:00 +13:00
}
2021-09-02 07:58:39 +12:00
manager.sessions.Broadcast(event.CONTROL_HOST, payload, nil)
2021-08-30 04:23:58 +12:00
manager.logger.Debug().
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
})
2021-02-15 05:11:21 +13:00
manager.desktop.OnClipboardUpdated(func() {
session := manager.sessions.GetHost()
2021-03-14 12:45:51 +13:00
if session == nil || !session.Profile().CanAccessClipboard {
2021-01-12 03:30:53 +13:00
return
}
2021-09-02 07:58:39 +12:00
manager.logger.Debug().Msg("sync clipboard")
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
}
2021-09-02 07:58:39 +12:00
session.Send(
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
2021-02-15 05:11:21 +13:00
manager.fileChooserDialogEvents()
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)
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-09-03 06:30:50 +12:00
// create new peer
peer := newPeer(connection)
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")
2021-09-03 07:52:23 +12:00
peer.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()
2021-09-03 06:30:50 +12:00
peer.setSessionID(session.ID())
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
if !manager.sessions.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
}
session.SetWebSocketPeer(peer)
2020-11-02 04:09:48 +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
2021-03-24 03:38:37 +13:00
session.SetWebSocketConnected(peer, true)
2020-11-26 08:26:11 +13:00
2020-11-02 04:09:48 +13:00
defer func() {
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 ended")
2020-11-26 08:26:11 +13:00
2021-03-24 03:38:37 +13:00
session.SetWebSocketConnected(peer, false)
2020-11-02 04:09:48 +13:00
}()
2021-09-25 01:15:59 +12:00
manager.handle(connection, peer, session)
2020-11-02 04:09:48 +13:00
}
2021-09-25 01:15:59 +12:00
func (manager *WebSocketManagerCtx) handle(connection *websocket.Conn, peer types.WebSocketPeer, session types.Session) {
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 struct{})
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 {
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
2021-08-30 04:23:58 +12:00
logger.Warn().Err(err).Msg("read message error")
2020-11-02 04:37:00 +13:00
} else {
2021-08-30 04:23:58 +12:00
logger.Debug().Err(err).Msg("read message error")
2020-11-02 04:37:00 +13:00
}
2021-02-03 06:28:32 +13:00
2020-11-02 04:37:00 +13:00
close(cancel)
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
}
2021-09-10 08:51:24 +12:00
if data.Event != event.SYSTEM_LOGS {
logger.Debug().
Str("address", connection.RemoteAddr().String()).
Str("event", data.Event).
Str("payload", string(data.Payload)).
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 <-cancel:
return
case <-manager.shutdown:
2021-09-25 01:15:59 +12:00
peer.Destroy("connection shutdown")
2021-09-19 00:59:15 +12:00
return
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 {
2021-09-02 09:10:06 +12:00
logger.Err(err).Msg("ping message has failed")
2020-11-02 04:09:48 +13:00
return
}
}
}
}