Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
155 changes: 155 additions & 0 deletions api/websh/idle_warning_test.go
Original file line number Diff line number Diff line change
@@ -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
}
19 changes: 11 additions & 8 deletions api/websh/types.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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
Expand Down
80 changes: 56 additions & 24 deletions api/websh/websh.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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",
Expand Down Expand Up @@ -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{}),
}
}

Expand All @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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
}
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down
Loading
Loading