diff --git a/api/websh/idle_warning_test.go b/api/websh/idle_warning_test.go new file mode 100644 index 00000000..1ab853c9 --- /dev/null +++ b/api/websh/idle_warning_test.go @@ -0,0 +1,155 @@ +package websh + +import ( + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/alpacax/alpacon-cli/pkg/testutil" + "github.com/gorilla/websocket" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestReadFromServer_IdleWarningUsesStderrAndLocalTime(t *testing.T) { + local := time.Local + time.Local = time.FixedZone("test", 9*60*60) + t.Cleanup(func() { time.Local = local }) + closesAt := time.Date(time.Now().Year()+1, time.January, 2, 6, 4, 5, 0, time.UTC) + + _, stdout, stderr := readIdleFrames(t, idleWarningFrame(closesAt)) + + assert.Empty(t, stdout) + assert.Equal(t, "[Idle: this session disconnects at 15:04:05 without input.]\r\n", stderr) +} + +func TestReadFromServer_IdleWarningPreservesReplay(t *testing.T) { + closesAt := time.Date(time.Now().Year()+1, time.January, 2, 15, 4, 5, 0, time.Local) + warning := idleWarningFrame(closesAt) + for _, tc := range []struct { + name string + frames []replayFrame + epoch string + offset uint64 + output bool + }{ + {"before origin", []replayFrame{warning, {websocket.TextMessage, `{"type":"replay_origin","epoch":"epoch-1","offset":100}`}, {websocket.BinaryMessage, "hello"}}, "epoch-1", 105, true}, + {"after origin", []replayFrame{{websocket.TextMessage, `{"type":"replay_origin","epoch":"epoch-1","offset":100}`}, warning, {websocket.BinaryMessage, "hello"}}, "epoch-1", 105, true}, + {"without binary", []replayFrame{warning}, "", 0, false}, + {"legacy", []replayFrame{{websocket.BinaryMessage, "hello"}, warning, {websocket.TextMessage, `{"type":"replay_origin","epoch":"too-late","offset":999}`}}, "", 0, true}, + } { + t.Run(tc.name, func(t *testing.T) { + conn, stdout, stderr := readIdleFrames(t, tc.frames...) + assert.Equal(t, tc.epoch, conn.replay.epoch) + assert.Equal(t, tc.offset, conn.replay.offset) + assert.Equal(t, tc.output, conn.replay.outputWritten) + if tc.output { + assert.Equal(t, "hello", stdout) + } else { + assert.Empty(t, stdout) + } + assert.Equal(t, "[Idle: this session disconnects at 15:04:05 without input.]\r\n", stderr) + }) + } +} + +func TestReadFromServer_IdleWarningIgnoresInvalidDeadlines(t *testing.T) { + closesAt := time.Date(time.Now().Year()+1, time.January, 2, 15, 4, 5, 0, time.Local) + for _, tc := range []struct { + name string + body string + }{ + {"missing", `{"type":"idle_warning"}`}, + {"null", `{"type":"idle_warning","closes_at":null}`}, + {"invalid", `{"type":"idle_warning","closes_at":"not-a-time"}`}, + {"past", idleWarningFrame(time.Now().Add(-time.Hour)).body}, + {"unknown type", `{"type":"future_control","closes_at":"` + closesAt.Format(time.RFC3339) + `"}`}, + } { + t.Run(tc.name, func(t *testing.T) { + _, stdout, stderr := readIdleFrames(t, + replayFrame{websocket.TextMessage, tc.body}, idleWarningFrame(closesAt), + ) + assert.Empty(t, stdout) + assert.Equal(t, "[Idle: this session disconnects at 15:04:05 without input.]\r\n", stderr) + }) + } +} + +func TestReadFromServer_IdleWarningPrintsOncePerDeadline(t *testing.T) { + first := time.Date(time.Now().Year()+1, time.January, 2, 15, 4, 5, 0, time.Local) + second := first.Add(time.Minute) + _, stdout, stderr := readIdleFrames(t, + idleWarningFrame(first), idleWarningFrame(first), idleWarningFrame(second), idleWarningFrame(first), + ) + assert.Empty(t, stdout) + assert.Equal(t, "[Idle: this session disconnects at 15:04:05 without input.]\r\n"+ + "[Idle: this session disconnects at 15:05:05 without input.]\r\n", stderr) +} + +func TestReadFromServer_IdleWarningStaysDeduplicatedOnReconnect(t *testing.T) { + closesAt := time.Date(time.Now().Year()+1, time.January, 2, 15, 4, 5, 0, time.Local) + _, stdout, notices := runReplaySession(t, + replayRound{frames: []replayFrame{idleWarningFrame(closesAt)}, closeCode: websocket.CloseServiceRestart}, + replayRound{frames: []replayFrame{idleWarningFrame(closesAt)}}, + ) + assert.Empty(t, stdout) + assert.Equal(t, "[Idle: this session disconnects at 15:04:05 without input.]\r\n\r\nConnection lost, reconnecting...\r\n", notices) +} + +func TestReadFromServer_IdleWarningPrunesExpiredHistory(t *testing.T) { + expired := time.Now().Add(-time.Hour).UTC() + retained := time.Now().Add(time.Hour).UTC().Truncate(time.Second) + next := retained.Add(time.Minute) + history := map[time.Time]struct{}{expired: {}, retained: {}} + + conn, stdout, stderr := readIdleFramesWithHistory(t, history, + idleWarningFrame(retained), idleWarningFrame(next), + ) + + assert.Empty(t, stdout) + assert.Equal(t, fmt.Sprintf("[Idle: this session disconnects at %s without input.]\r\n", next.Local().Format("15:04:05")), stderr) + assert.NotContains(t, conn.idleWarnings, expired) + assert.Contains(t, conn.idleWarnings, retained) + assert.Contains(t, conn.idleWarnings, next) + assert.Len(t, conn.idleWarnings, 2) +} + +func idleWarningFrame(closesAt time.Time) replayFrame { + return replayFrame{websocket.TextMessage, fmt.Sprintf(`{"type":"idle_warning","closes_at":%q,"message":"untrusted server text"}`, closesAt.Format(time.RFC3339))} +} + +func readIdleFrames(t *testing.T, frames ...replayFrame) (*connection, string, string) { + t.Helper() + return readIdleFramesWithHistory(t, nil, frames...) +} + +func readIdleFramesWithHistory(t *testing.T, history map[time.Time]struct{}, frames ...replayFrame) (*connection, string, string) { + t.Helper() + upgrader := websocket.Upgrader{} + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ws, err := upgrader.Upgrade(w, r, nil) + if !assert.NoError(t, err) { + return + } + defer func() { _ = ws.Close() }() + for _, frame := range frames { + if !assert.NoError(t, ws.WriteMessage(frame.kind, []byte(frame.body))) { + return + } + } + _ = ws.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "")) + })) + t.Cleanup(ts.Close) + ws, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(ts.URL, "http"), nil) + require.NoError(t, err) + conn := newConnection(ws) + conn.idleWarnings = history + t.Cleanup(conn.close) + stdout, stderr := testutil.CaptureOutput(t, func() { + awaitReturn(t, "warning reader did not finish", conn.readFromServer) + }) + return conn, stdout, stderr +} diff --git a/api/websh/types.go b/api/websh/types.go index 119d2ee8..362e3a7f 100644 --- a/api/websh/types.go +++ b/api/websh/types.go @@ -23,6 +23,8 @@ type WebsocketClient struct { reconnect *reconnector // The reader owns replay state until its pumps stop before the next dial. replay replayState + // The reader keeps warning deadlines across connections, like replay state. + idleWarnings map[time.Time]struct{} } type replayState struct { @@ -35,14 +37,15 @@ type replayState struct { // connections—a dropped link is re-dialed onto a new user channel—so each one // records its own ending and the session decides what that ending means. type connection struct { - ws *websocket.Conn - ended chan struct{} // closed once the first ending is recorded - err error - endOnce sync.Once - pumps sync.WaitGroup - replay *replayState - reconnected bool - notice io.Writer + ws *websocket.Conn + ended chan struct{} // closed once the first ending is recorded + err error + endOnce sync.Once + pumps sync.WaitGroup + replay *replayState + reconnected bool + notice io.Writer + idleWarnings map[time.Time]struct{} } // reconnector is what an interactive session needs to survive a dropped diff --git a/api/websh/websh.go b/api/websh/websh.go index 5bd97ad1..5d74e906 100644 --- a/api/websh/websh.go +++ b/api/websh/websh.go @@ -30,15 +30,10 @@ const ( ctrlC = 0x03 writeFlushInterval = 5 * time.Millisecond - // sessionEndCloseCode is how the proxy, and Alpamon, close a user channel at the - // end of a session. A websh session never ends with 1000, so without it every - // normal end would be reported as a failure. + // sessionEndCloseCode treats 4000 as normal; the server never ends sessions with 1000. sessionEndCloseCode = 4000 - // A close that does not end the session is a dropped link (a restarted service, a - // network interruption); the session still runs on the server, so the client dials - // a new channel onto it. The budget, about 30s of waiting, bounds a server that - // keeps accepting a channel and dropping it. + // 1012 and 1006 leave the session running, so re-dial; cap repeated drops. maxReconnectAttempts = 5 reconnectBaseDelay = 1 * time.Second reconnectMaxDelay = 15 * time.Second @@ -61,12 +56,13 @@ var ( // closed while the connection was down, so retrying only asks again. ErrSessionGone = errors.New("the session is no longer open") - // closeReasonMessages maps the proxy's session-end reason tokens; an empty message stays + // closeReasonMessages maps the proxy's close reason tokens; an empty message stays // silent, since shell_exited means the user ended the shell and closed says nothing. closeReasonMessages = map[string]string{ "user_request": "the session was closed", "force_close": "closed by an administrator", - "idle_timeout": "the session reached its time limit", + "idle_timeout": "after inactivity", + "idle_detached": "after inactivity; the shell keeps running on the server until the work session ends", "agent_disconnected": "Alpamon on the server disconnected", "server_shutdown": "the service restarted", "work_session_expired": "the work session expired", @@ -316,8 +312,9 @@ func shareCreatedSession(ac *client.AlpaconClient, responseBody []byte, share, r func newWebsocketClient(header http.Header) *WebsocketClient { return &WebsocketClient{ - header: header, - done: make(chan struct{}), + header: header, + done: make(chan struct{}), + idleWarnings: make(map[time.Time]struct{}), } } @@ -340,6 +337,7 @@ func (wsClient *WebsocketClient) dial(websocketURL string) error { } wsClient.conn = newConnection(conn) wsClient.conn.replay = &wsClient.replay + wsClient.conn.idleWarnings = wsClient.idleWarnings if wsClient.reconnect != nil { wsClient.conn.reconnected = reconnected wsClient.conn.notice = wsClient.reconnect.notice @@ -419,14 +417,18 @@ func printCloseReason(err error) { if message == "" { return } - _, _ = fmt.Fprintf(os.Stderr, "\r\nsession closed: %s\r\n", message) + prefix := "session closed: " + switch reason { + case "idle_timeout": + prefix = "session closed " + case "idle_detached": + prefix = "session disconnected " + } + _, _ = fmt.Fprintf(os.Stderr, "\r\n%s%s\r\n", prefix, message) } } -// endsSession reports whether a connection error means the session itself ended: -// the proxy's session-end code or a normal WebSocket close. Any other ending (1012, -// 1006, a transport error with no close frame) leaves the session running on the -// server, so the connection is re-dialed. +// endsSession treats 4000 as final without reconnecting, even when idle detach leaves the shell running. func endsSession(err error) bool { if err == nil { return true @@ -451,7 +453,7 @@ func reconnectDelay(failures int, base time.Duration) time.Duration { // Input is not forwarded to the server. Terminal echo is suppressed via raw mode. // Ends cleanly on the remote close, on Ctrl+C, or on a signal. func OpenReadOnlyTerminal(ac *client.AlpaconClient, sessionResponse SessionResponse) error { - wsClient := newWebsocketClient(ac.SetWebsocketHeader()) + wsClient := newWebsocketClient(ac.SetWebsocketHeaderWithCapabilities(client.CapabilityIdleWarning)) if err := wsClient.dial(sessionResponse.WebsocketURL); err != nil { return err } @@ -528,9 +530,9 @@ func OpenSharedTerminal(ac *client.AlpaconClient, sessionResponse SessionRespons func openInteractiveTerminal(ac *client.AlpaconClient, sessionResponse SessionResponse, reconnect *reconnector) error { var header http.Header if reconnect != nil { - header = ac.SetWebsocketHeaderWithCapabilities(client.CapabilityWebsocketReconnect) + header = ac.SetWebsocketHeaderWithCapabilities(client.CapabilityWebsocketReconnect, client.CapabilityIdleWarning) } else { - header = ac.SetWebsocketHeader() + header = ac.SetWebsocketHeaderWithCapabilities(client.CapabilityIdleWarning) } wsClient := newWebsocketClient(header) @@ -706,12 +708,42 @@ func (c *connection) readFromServer() { switch kind { case websocket.TextMessage: var origin struct { - Type string `json:"type"` - Epoch string `json:"epoch"` - Offset uint64 `json:"offset"` - Truncated bool `json:"truncated"` + Type string `json:"type"` + Epoch string `json:"epoch"` + Offset uint64 `json:"offset"` + Truncated bool `json:"truncated"` + ClosesAt json.RawMessage `json:"closes_at"` + } + if json.Unmarshal(message, &origin) != nil { + continue + } + if origin.Type == "idle_warning" { + var closesAt time.Time + now := time.Now() + if err := json.Unmarshal(origin.ClosesAt, &closesAt); err != nil || !closesAt.After(now) { + continue + } + for deadline := range c.idleWarnings { + if !deadline.After(now) { + delete(c.idleWarnings, deadline) + } + } + closesAt = closesAt.UTC() + if _, seen := c.idleWarnings[closesAt]; seen { + continue + } + if c.idleWarnings == nil { + c.idleWarnings = make(map[time.Time]struct{}) + } + c.idleWarnings[closesAt] = struct{}{} + notice := c.notice + if notice == nil { + notice = os.Stderr + } + _, _ = fmt.Fprintf(notice, "[Idle: this session disconnects at %s without input.]\r\n", closesAt.Local().Format("15:04:05")) + continue } - if legacy || json.Unmarshal(message, &origin) != nil || origin.Type != "replay_origin" || origin.Epoch == "" { + if legacy || origin.Type != "replay_origin" || origin.Epoch == "" { continue } originSeen = true diff --git a/api/websh/websh_test.go b/api/websh/websh_test.go index dfef4cd3..c50025c1 100644 --- a/api/websh/websh_test.go +++ b/api/websh/websh_test.go @@ -586,9 +586,14 @@ func TestFinish_SettlesASingleCloseByItsCode(t *testing.T) { keepAsErr bool }{ { - name: "session end with a reason", + name: "idle timeout closes the shell", reported: &websocket.CloseError{Code: sessionEndCloseCode, Text: "idle_timeout"}, - wantStderr: "\r\nsession closed: the session reached its time limit\r\n", + wantStderr: "\r\nsession closed after inactivity\r\n", + }, + { + name: "idle detach keeps the shell running", + reported: &websocket.CloseError{Code: sessionEndCloseCode, Text: "idle_detached"}, + wantStderr: "\r\nsession disconnected after inactivity; the shell keeps running on the server until the work session ends\r\n", }, { name: "session end with no reason", @@ -725,14 +730,14 @@ func TestFinish_KeepsTheFirstOutcome(t *testing.T) { { name: "earlier failure over a later session end", first: errors.New("write failed on the closed connection"), - second: &websocket.CloseError{Code: sessionEndCloseCode, Text: "idle_timeout"}, + second: &websocket.CloseError{Code: sessionEndCloseCode, Text: "idle_detached"}, wantErr: errors.New("write failed on the closed connection"), }, { name: "duplicate session end prints once", - first: &websocket.CloseError{Code: sessionEndCloseCode, Text: "idle_timeout"}, - second: &websocket.CloseError{Code: sessionEndCloseCode, Text: "idle_timeout"}, - wantStderr: "\r\nsession closed: the session reached its time limit\r\n", + first: &websocket.CloseError{Code: sessionEndCloseCode, Text: "idle_detached"}, + second: &websocket.CloseError{Code: sessionEndCloseCode, Text: "idle_detached"}, + wantStderr: "\r\nsession disconnected after inactivity; the shell keeps running on the server until the work session ends\r\n", }, } @@ -1104,7 +1109,7 @@ func TestServeConnections_ReconnectsAfterAServiceRestart(t *testing.T) { t.Cleanup(ts.Close) ac := &client.AlpaconClient{HTTPClient: ts.Client(), BaseURL: ts.URL, UserAgent: "alpacon-cli/test"} - wsClient := newWebsocketClient(ac.SetWebsocketHeaderWithCapabilities(client.CapabilityWebsocketReconnect)) + wsClient := newWebsocketClient(ac.SetWebsocketHeaderWithCapabilities(client.CapabilityWebsocketReconnect, client.CapabilityIdleWarning)) wsClient.reconnect = newReconnector(ac, "sess-1") wsClient.reconnect.baseDelay = time.Millisecond wsClient.reconnect.notice = io.Discard @@ -1143,7 +1148,7 @@ func TestServeConnections_ReconnectsAfterAServiceRestart(t *testing.T) { for dial := 1; dial <= 2; dial++ { select { case header := <-dialHeaders: - assert.Equal(t, client.CapabilityWebsocketReconnect, + assert.Equal(t, "websocket-reconnect, idle-warning", header.Get(client.ClientCapabilitiesHeader), "dial %d", dial) assert.Equal(t, "alpacon-cli/test", header.Get("User-Agent"), "dial %d", dial) default: @@ -1158,6 +1163,49 @@ func TestServeConnections_ReconnectsAfterAServiceRestart(t *testing.T) { } } +func TestServeConnections_IdleDetachDoesNotReconnect(t *testing.T) { + upgrader := websocket.Upgrader{} + var dials atomic.Int32 + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + dials.Add(1) + ws, err := upgrader.Upgrade(w, r, nil) + if !assert.NoError(t, err) { + return + } + defer func() { _ = ws.Close() }() + assert.NoError(t, ws.WriteMessage(websocket.CloseMessage, + websocket.FormatCloseMessage(sessionEndCloseCode, "idle_detached"))) + })) + t.Cleanup(ts.Close) + + wsClient := newWebsocketClient(nil) + var attempts int + var notices strings.Builder + wsClient.reconnect = &reconnector{ + provision: func() (string, error) { + attempts++ + return "", errors.New("unexpected reconnect after idle detach") + }, + notice: ¬ices, baseDelay: time.Millisecond, maxAttempts: 1, + } + require.NoError(t, wsClient.dial("ws"+strings.TrimPrefix(ts.URL, "http"))) + t.Cleanup(wsClient.closeConn) + + stdout, stderr := testutil.CaptureOutput(t, func() { + awaitReturn(t, "idle detach did not end the client", func() { + wsClient.serveConnections(make(chan string)) + }) + }) + + assertReported(t, wsClient) + require.NoError(t, wsClient.err) + assert.Zero(t, attempts) + assert.Equal(t, int32(1), dials.Load()) + assert.Empty(t, stdout) + assert.Empty(t, notices.String()) + assert.Equal(t, "\r\nsession disconnected after inactivity; the shell keeps running on the server until the work session ends\r\n", stderr) +} + func TestRedial_StopsWhenTheServerRefusesANewChannel(t *testing.T) { t.Parallel() var attempts atomic.Int32 @@ -1245,7 +1293,7 @@ func pinTerminalSize(t *testing.T, rows, cols int) { // What reaches the server on the real entry points' first dial. Raw mode fails on // a pipe, so each call returns right after the handshake the assertions read. -func TestOpenTerminal_AdvertisesTheCapabilityOnlyWhereItReconnects(t *testing.T) { +func TestOpenTerminal_AdvertisesIdleWarningAndReconnectCapabilities(t *testing.T) { tests := []struct { name string open func(*client.AlpaconClient, SessionResponse) error @@ -1254,13 +1302,19 @@ func TestOpenTerminal_AdvertisesTheCapabilityOnlyWhereItReconnects(t *testing.T) { name: "an interactive session reconnects", open: OpenNewTerminal, - want: client.CapabilityWebsocketReconnect, + want: "websocket-reconnect, idle-warning", }, { // The joiner may not open a channel on someone else's session, so - // claiming the capability would promise a recovery it cannot make. - name: "a joined session does not", + // claiming the reconnect capability would promise a recovery it cannot make. + name: "a joined session warns without reconnecting", open: OpenSharedTerminal, + want: "idle-warning", + }, + { + name: "a watched session warns without reconnecting", + open: OpenReadOnlyTerminal, + want: "idle-warning", }, } diff --git a/client/client.go b/client/client.go index d0a71bb9..bd270b77 100644 --- a/client/client.go +++ b/client/client.go @@ -37,6 +37,9 @@ const ( // for a transient reason does not take the session with it. CapabilityWebsocketReconnect = "websocket-reconnect" + // CapabilityIdleWarning says the client handles idle warnings without printing raw frames. + CapabilityIdleWarning = "idle-warning" + // serverGateTokenScope is the "gate" of a token-scope refusal, the only gate whose "missing" is // a scope string rather than a permission or role. serverGateTokenScope = "token_scope"