From 47f1b2e55b2bb8f31c663e2ba793b7e65047a868 Mon Sep 17 00:00:00 2001 From: iRedTea Date: Wed, 14 Oct 2026 21:06:32 +0300 Subject: [PATCH] feat: the browser terminal on a WebSocket too, long polls when it does not connect or drops --- internal/server/handlers_panel_term.go | 78 +++++++++++++++++++++++++- internal/server/hub.go | 8 +++ internal/server/server.go | 1 + internal/server/web/static/terminal.js | 52 +++++++++++++---- 4 files changed, 127 insertions(+), 12 deletions(-) diff --git a/internal/server/handlers_panel_term.go b/internal/server/handlers_panel_term.go index 474d733..8dc6415 100644 --- a/internal/server/handlers_panel_term.go +++ b/internal/server/handlers_panel_term.go @@ -1,9 +1,14 @@ package server import ( + "context" "encoding/json" "log" "net/http" + "time" + + "github.com/coder/websocket" + "github.com/coder/websocket/wsjson" "github.com/x-grk/longpolling-remote-access/internal/proto" ) @@ -66,8 +71,14 @@ func (s *Server) handlePanelIn(w http.ResponseWriter, r *http.Request) { http.Error(w, "bad request", http.StatusBadRequest) return } + takePanelIn(sess, m) + w.WriteHeader(http.StatusNoContent) +} + +// takePanelIn passes the browser's keystrokes or resize on to the agent; a +// repeat of one already taken is dropped. +func takePanelIn(sess *Session, m panelInMsg) { if m.Seq > 0 && !sess.TakeSeq(m.Seq) { - w.WriteHeader(http.StatusNoContent) // a repeat of one already taken return } switch m.T { @@ -76,7 +87,70 @@ func (s *Server) handlePanelIn(w http.ResponseWriter, r *http.Request) { case "r": sess.PushIn(proto.InFrame{Type: proto.FrameResize, Cols: m.Cols, Rows: m.Rows}) } - w.WriteHeader(http.StatusNoContent) +} + +// GET /api/panel/term/{sid}/ws - the browser's terminal on a WebSocket: what +// the "in" requests send comes as messages, and output goes out as soon as +// it is there, in the "out" answers' shape. The page falls back to those +// requests when this does not connect. +func (s *Server) handlePanelTermWS(w http.ResponseWriter, r *http.Request) { + sess := s.hub.Session(r.PathValue("sid")) + if sess == nil { + http.Error(w, "closed", http.StatusGone) + return + } + conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{CompressionMode: websocket.CompressionDisabled}) + if err != nil { + return + } + conn.SetReadLimit(1 << 20) + defer conn.CloseNow() + ctx, cancel := context.WithCancel(r.Context()) + defer cancel() + go func() { + defer cancel() + for ctx.Err() == nil { + msgs, alive := sess.PollOutUntil(ctx.Done(), pollHold) + if ctx.Err() != nil { + return + } + out := map[string]any{"frames": msgs} + if !alive { + out = map[string]any{"closed": true} + } + wctx, wcancel := context.WithTimeout(ctx, 30*time.Second) + err := wsjson.Write(wctx, conn, out) + wcancel() + if err != nil || !alive { + return + } + } + }() + go func() { + t := time.NewTicker(wsPing) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + pctx, pcancel := context.WithTimeout(ctx, 15*time.Second) + err := conn.Ping(pctx) + pcancel() + if err != nil { + cancel() + return + } + } + } + }() + for { + var m panelInMsg + if err := wsjson.Read(ctx, conn, &m); err != nil { + return + } + takePanelIn(sess, m) + } } // POST /api/panel/term/{sid}/close - tear the session down. diff --git a/internal/server/hub.go b/internal/server/hub.go index b6ab05e..8c2df3c 100644 --- a/internal/server/hub.go +++ b/internal/server/hub.go @@ -324,10 +324,18 @@ func (s *Session) Out() <-chan proto.OutMessage { return s.out } // PollOut drains queued upstream messages, blocking up to timeout for the first. func (s *Session) PollOut(timeout time.Duration) ([]proto.OutMessage, bool) { + return s.PollOutUntil(nil, timeout) +} + +// PollOutUntil is PollOut that also gives up when done closes, taking +// nothing: for a reader that has gone, so no output is lost on it. +func (s *Session) PollOutUntil(done <-chan struct{}, timeout time.Duration) ([]proto.OutMessage, bool) { var batch []proto.OutMessage select { case m := <-s.out: batch = append(batch, m) + case <-done: + return nil, true case <-time.After(timeout): return nil, true case <-s.closed: diff --git a/internal/server/server.go b/internal/server/server.go index 1ec53a4..8aa6984 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -249,6 +249,7 @@ func (s *Server) Handler() http.Handler { mux.Handle("POST /api/panel/term/{agentID}/open", s.requireAdmin(s.handlePanelOpen)) mux.Handle("GET /api/panel/term/{sid}/out", s.requireAdmin(s.handlePanelOut)) mux.Handle("POST /api/panel/term/{sid}/in", s.requireAdmin(s.handlePanelIn)) + mux.Handle("GET /api/panel/term/{sid}/ws", s.requireAdmin(s.handlePanelTermWS)) mux.Handle("POST /api/panel/term/{sid}/close", s.requireAdmin(s.handlePanelClose)) // Panel file manager (admin-only). diff --git a/internal/server/web/static/terminal.js b/internal/server/web/static/terminal.js index ab72a47..c89168c 100644 --- a/internal/server/web/static/terminal.js +++ b/internal/server/web/static/terminal.js @@ -76,13 +76,47 @@ this.sid = (await r.json()).session_id; this.setState('wait', 'ожидание агента…'); this.fit(); + // A WebSocket if it connects; the long polls otherwise, and from a + // dropped WebSocket on for the rest of this session. + if (!(await this.openWS())) this.poll(); this.queueIn({ t: 'r', cols: this.term.cols, rows: this.term.rows }); - this.poll(); } catch (e) { this.setState('err', 'ошибка сети'); this.term.write('\r\n\x1b[31m[ошибка сети: ' + (e && e.message ? e.message : e) + ']\x1b[0m\r\n'); } } + openWS() { + return new Promise(resolve => { + if (!window.WebSocket) { resolve(false); return; } + const ws = new WebSocket((location.protocol === 'https:' ? 'wss://' : 'ws://') + location.host + base + this.sid + '/ws'); + let opened = false; + const timer = setTimeout(() => { if (!opened) { ws.close(); resolve(false); } }, 5000); + ws.onopen = () => { opened = true; clearTimeout(timer); this.ws = ws; resolve(true); }; + ws.onmessage = ev => { + let j; + try { j = JSON.parse(ev.data); } catch (e) { return; } + this.take(j); + }; + ws.onclose = () => { + clearTimeout(timer); + if (!opened) { resolve(false); return; } + this.ws = null; + if (!this.closed && !this.ended) this.poll(); + }; + }); + } + // take applies one answer of the output channel, either transport's. + take(j) { + if (j.closed) { this.finish('[сессия завершена]'); return false; } + for (const f of j.frames || []) { + if (f.exit) { this.finish('[сессия завершена]'); return false; } + if (f.data) { + if (!this.connected) { this.connected = true; this.setState('ok', 'подключено'); } + this.term.write(b64ToBytes(f.data)); + } + } + return true; + } async poll() { while (!this.closed && this.sid) { let r; @@ -90,15 +124,7 @@ catch (e) { await new Promise(res => setTimeout(res, 1000)); continue; } if (this.closed) return; if (!r.ok) { this.finish('[соединение закрыто]'); return; } - const j = await r.json(); - if (j.closed) { this.finish('[сессия завершена]'); return; } - for (const f of j.frames || []) { - if (f.exit) { this.finish('[сессия завершена]'); return; } - if (f.data) { - if (!this.connected) { this.connected = true; this.setState('ok', 'подключено'); } - this.term.write(b64ToBytes(f.data)); - } - } + if (!this.take(await r.json())) return; } } finish(msg) { @@ -126,6 +152,11 @@ const body = m.t === 'i' ? { t: 'i', d: bytesToB64(new TextEncoder().encode(m.text)), seq: ++this.inSeq } : { t: 'r', cols: m.cols, rows: m.rows, seq: ++this.inSeq }; + // On the WebSocket a message goes out in order with no answer to wait + // for; numbered as well, so a repeat by the long polls is dropped. + if (this.ws && this.ws.readyState === 1) { + try { this.ws.send(JSON.stringify(body)); continue; } catch (e) { /* by request below */ } + } for (let wait = 300; !this.closed; wait = Math.min(wait * 2, 5000)) { let r = null; try { @@ -145,6 +176,7 @@ close() { if (this.closed) return; this.closed = true; + if (this.ws) { try { this.ws.close(); } catch (e) { /* closed already */ } } if (this.sid) fetch(base + this.sid + '/close', { method: 'POST', keepalive: true }).catch(() => {}); this.term.dispose(); this.el.remove();