From 57f689d15c2d93fc1b90c1c0ff86ec7a4e2921a5 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 21 Sep 2026 17:28:02 -0600 Subject: [PATCH] fix(ssh): bound tunnel client handshake timeout Bound existing-connection SSH handshakes by idle progress with a two-minute backstop, classify stalls as transient, and retry setup failures. Add coverage and SSH server startup diagnostics. Signed-off-by: Samuel K --- cmd/internal/ssh_server.go | 3 + e2e/framework/retry.go | 14 ++-- e2e/framework/retry_test.go | 50 +++++++++++++++ pkg/ssh/helper.go | 123 +++++++++++++++++++++++++++++++++++- pkg/ssh/helper_test.go | 123 ++++++++++++++++++++++++++++++++++++ 5 files changed, 307 insertions(+), 6 deletions(-) diff --git a/cmd/internal/ssh_server.go b/cmd/internal/ssh_server.go index 174fe3e38..906476ce4 100644 --- a/cmd/internal/ssh_server.go +++ b/cmd/internal/ssh_server.go @@ -87,6 +87,7 @@ func (cmd *sshServerCmd) run(ctx context.Context) error { // internal SSH cleanup could run. Liveness is decided via a per-directory // flock the owning process holds for its lifetime; the kernel releases // the flock on any process exit (including SIGKILL). + log.Debugf("starting ssh server: stdio=%v", cmd.stdio) sshserver.SweepStaleAgentSockets() server, err := sshserver.NewServer( @@ -95,6 +96,7 @@ func (cmd *sshServerCmd) run(ctx context.Context) error { if err != nil { return fmt.Errorf("create ssh server: %w", err) } + log.Debug("ssh server initialized") if cmd.stdio { return cmd.serveStdio(ctx, server) @@ -108,6 +110,7 @@ func (cmd *sshServerCmd) serveStdio(ctx context.Context, server sshserver.Server } go shutdownOnCancel(ctx, server) // #nosec G118 -- see shutdownOnCancel. lis := stdio.NewStdioListener(os.Stdin, os.Stdout) + log.Debug("serving ssh on stdio") return ignoreServerClosed(server.Serve(lis)) } diff --git a/e2e/framework/retry.go b/e2e/framework/retry.go index 62db435bf..4fd1308c5 100644 --- a/e2e/framework/retry.go +++ b/e2e/framework/retry.go @@ -53,6 +53,7 @@ var retryableSSHPatterns = []string{ "connection timed out", "broken pipe", "workspace not found", + "ssh handshake made no progress", } // isRetryableSSHError returns true when the error indicates a transient SSH @@ -91,8 +92,10 @@ func isRetryableDockerError(stderr string) bool { return false } -// execWithDockerRetry runs fn and retries if stderr indicates a transient -// Docker registry error. Returns the last stdout, stderr, and error. +// execWithDockerRetry runs fn and retries if the failure looks transient: +// either a Docker registry error or a retryable SSH setup error in stderr +// (up commands build the tunnel before the workspace starts). Returns the +// last stdout, stderr, and error. func execWithDockerRetry( ctx context.Context, fn func(ctx context.Context) (stdout, stderr string, err error), @@ -109,19 +112,20 @@ func execWithDockerRetry( if lastErr == nil { return lastStdout, lastStderr, nil } - if !isRetryableDockerError(lastStderr) || attempt == dockerPullBackoff.Steps { + retryable := isRetryableDockerError(lastStderr) || isRetryableSSHError(lastErr, lastStderr) + if !retryable || attempt == dockerPullBackoff.Steps { break } delay := nextBackoffDelay(dockerPullBackoff, attempt) if !retryFitsBudget(ctx, delay) { return lastStdout, lastStderr, fmt.Errorf( - "after %d attempts: retryable Docker error; retry not attempted because "+ + "after %d attempts: retryable transient error; retry not attempted because "+ "remaining deadline budget was insufficient (next retry delay: %s): %w", attempt, delay, lastErr, ) } ginkgo.GinkgoWriter.Printf( - "[retry] attempt %d failed with transient Docker error, retrying after %s: %s\n", + "[retry] attempt %d failed with transient error, retrying after %s: %s\n", attempt, delay, lastErr, ) if err := waitForRetry(ctx, delay); err != nil { diff --git a/e2e/framework/retry_test.go b/e2e/framework/retry_test.go index badb4a64c..c80085ecc 100644 --- a/e2e/framework/retry_test.go +++ b/e2e/framework/retry_test.go @@ -256,6 +256,37 @@ func TestExecWithDockerRetry_RetryThenSuccess(t *testing.T) { assert.Equal(t, "", stderr) } +func TestExecWithDockerRetry_HandshakeStallRetried(t *testing.T) { + withFastBackoffs(t) + calls := 0 + stderrMsg := "start workspace: failed to create SSH client: ssh handshake made no progress for 15s" + out, _, err := execWithDockerRetry(context.Background(), + func(context.Context) (string, string, error) { + calls++ + if calls == 1 { + return "", stderrMsg, transientExitErr(t) + } + return "ok", "", nil + }, + ) + require.NoError(t, err) + assert.Equal(t, 2, calls) + assert.Equal(t, "ok", out) +} + +func TestExecWithDockerRetry_NonRetryableExitNotRetried(t *testing.T) { + withFastBackoffs(t) + calls := 0 + _, _, err := execWithDockerRetry(context.Background(), + func(context.Context) (string, string, error) { + calls++ + return "", "provider \"docker123\" does not exist", transientExitErr(t) + }, + ) + require.Error(t, err) + assert.Equal(t, 1, calls) +} + func TestExecWithDockerRetry_RetryExhausted(t *testing.T) { withFastBackoffs(t) calls := 0 @@ -365,6 +396,25 @@ func TestExecWithSSHRetry_RetryThenSuccess(t *testing.T) { assert.Equal(t, "ok", out) } +func TestExecWithSSHRetry_RetryHandshakeStallThenSuccess(t *testing.T) { + withFastBackoffs(t) + calls := 0 + out, err := execWithSSHRetry(context.Background(), "ws", + func(context.Context) (string, string, error) { + calls++ + if calls == 1 { + return "", "run in container: ssh client: ssh handshake made no progress for 15s", transientExitErr( + t, + ) + } + return "ok", "", nil + }, + ) + require.NoError(t, err) + assert.Equal(t, 2, calls) + assert.Equal(t, "ok", out) +} + func TestExecWithSSHRetry_RetryExhausted(t *testing.T) { withFastBackoffs(t) calls := 0 diff --git a/pkg/ssh/helper.go b/pkg/ssh/helper.go index 6293e6391..ef4c7a664 100644 --- a/pkg/ssh/helper.go +++ b/pkg/ssh/helper.go @@ -6,6 +6,8 @@ import ( "fmt" "io" "net" + "sync/atomic" + "time" "github.com/devsy-org/devsy/pkg/stdio" "golang.org/x/crypto/ssh" @@ -90,8 +92,83 @@ func StdioClientFromKeyBytesWithUser( return ClientFromConn(conn, user, keyBytes) } +// HandshakeIdleTimeout fails a handshake that makes no progress for this +// long. Progress-based instead of absolute so slow-but-alive links (high +// latency is the norm for tunneled handshakes) always complete while a peer +// that stops sending entirely surfaces an error to callers. Zero disables. +const HandshakeIdleTimeout = 15 * time.Second + +// handshakeMaxTimeout caps a handshake whose peer never goes silent, e.g. +// one that dribbles a byte at a time. +const handshakeMaxTimeout = 2 * time.Minute + +// HandshakeTimeoutError reports a handshake that stopped making progress. +type HandshakeTimeoutError struct{ idle time.Duration } + +func (e *HandshakeTimeoutError) Error() string { + return fmt.Sprintf("ssh handshake made no progress for %s", e.idle) +} + +func (e *HandshakeTimeoutError) Timeout() bool { return true } + +// Temporary reports the error as transient so callers with net.Error retry +// handling treat it as worth retrying. +func (e *HandshakeTimeoutError) Temporary() bool { return true } + // ClientFromConn creates an SSH client over an existing network connection. +// The handshake is bounded by HandshakeIdleTimeout so a stalled peer surfaces +// an error instead of blocking the tunnel forever. No deadlines are set on +// the conn, so the established session is unaffected. func ClientFromConn(conn net.Conn, user string, keyBytes []byte) (*ssh.Client, error) { + return clientFromConn(conn, user, keyBytes, HandshakeIdleTimeout) +} + +// activityConn records the time of the last successful read or write. The +// stored time keeps its monotonic reading so idle measurement is immune to +// wall-clock corrections. +type activityConn struct { + net.Conn + tracking atomic.Bool + lastActivity atomic.Pointer[time.Time] +} + +func (c *activityConn) Read(b []byte) (int, error) { + n, err := c.Conn.Read(b) + if n > 0 { + c.recordActivity() + } + return n, err +} + +func (c *activityConn) Write(b []byte) (int, error) { + n, err := c.Conn.Write(b) + if n > 0 { + c.recordActivity() + } + return n, err +} + +func (c *activityConn) recordActivity() { + if !c.tracking.Load() { + return + } + now := time.Now() + c.lastActivity.Store(&now) +} + +type handshakeResult struct { + c ssh.Conn + chans <-chan ssh.NewChannel + reqs <-chan *ssh.Request + err error +} + +func clientFromConn( + conn net.Conn, + user string, + keyBytes []byte, + handshakeIdleTimeout time.Duration, +) (*ssh.Client, error) { if conn == nil { return nil, fmt.Errorf("connection is required") } @@ -101,14 +178,58 @@ func ClientFromConn(conn net.Conn, user string, keyBytes []byte) (*ssh.Client, e } clientConfig.User = user + if handshakeIdleTimeout > 0 { + return handshakeWithIdleTimeout(conn, clientConfig, handshakeIdleTimeout) + } c, chans, req, err := ssh.NewClientConn(conn, "stdio", clientConfig) if err != nil { return nil, err } - return ssh.NewClient(c, chans, handleKeepAliveRequests(req)), nil } +// handshakeWithIdleTimeout runs the handshake in a goroutine and closes the +// conn when it stops making progress, so stall detection works on conns that +// do not support deadlines. +func handshakeWithIdleTimeout( + conn net.Conn, + clientConfig *ssh.ClientConfig, + idle time.Duration, +) (*ssh.Client, error) { + tracked := &activityConn{Conn: conn} + tracked.tracking.Store(true) + tracked.recordActivity() + result := make(chan handshakeResult, 1) + go func() { + c, chans, req, err := ssh.NewClientConn(tracked, "stdio", clientConfig) + result <- handshakeResult{c: c, chans: chans, reqs: req, err: err} + }() + + ticker := time.NewTicker(idle / 4) + defer ticker.Stop() + deadline := time.After(handshakeMaxTimeout) + for { + select { + case res := <-result: + tracked.tracking.Store(false) + if res.err != nil { + return nil, res.err + } + return ssh.NewClient(res.c, res.chans, handleKeepAliveRequests(res.reqs)), nil + case <-ticker.C: + if time.Since(*tracked.lastActivity.Load()) > idle { + tracked.tracking.Store(false) + _ = conn.Close() + return nil, &HandshakeTimeoutError{idle: idle} + } + case <-deadline: + tracked.tracking.Store(false) + _ = conn.Close() + return nil, &HandshakeTimeoutError{idle: handshakeMaxTimeout} + } + } +} + func ConfigFromKeyBytes(keyBytes []byte) (*ssh.ClientConfig, error) { clientConfig := &ssh.ClientConfig{ Auth: []ssh.AuthMethod{}, diff --git a/pkg/ssh/helper_test.go b/pkg/ssh/helper_test.go index 6412e6aaa..2fa6867bc 100644 --- a/pkg/ssh/helper_test.go +++ b/pkg/ssh/helper_test.go @@ -2,9 +2,14 @@ package ssh import ( "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" "errors" "io" + "net" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -123,3 +128,121 @@ func TestSetupContextCancellation_ReturnsCleanupThatStopsWatcher(t *testing.T) { assert.NotPanics(t, cleanup) } + +type deadlineRecorder struct { + net.Conn + deadlines []time.Time +} + +func (d *deadlineRecorder) SetDeadline(t time.Time) error { + d.deadlines = append(d.deadlines, t) + return d.Conn.SetDeadline(t) +} + +func TestClientFromConn_StalledHandshakeReturnsError(t *testing.T) { + clientEnd, serverEnd := net.Pipe() + defer func() { _ = serverEnd.Close() }() + + conn := &deadlineRecorder{Conn: clientEnd} + start := time.Now() + client, err := clientFromConn(conn, "", nil, 50*time.Millisecond) + + require.Error(t, err) + assert.Nil(t, client) + assert.Less(t, time.Since(start), 5*time.Second) + var netErr net.Error + require.ErrorAs(t, err, &netErr) + assert.True(t, netErr.Timeout()) + assert.Empty(t, conn.deadlines, "handshake bound must not set conn deadlines") + + // the stalled conn is closed so the peer and any blocked goroutine unwind + _, werr := clientEnd.Write([]byte("x")) + require.Error(t, werr) +} + +func TestClientFromConn_SuccessfulHandshake(t *testing.T) { + // net.Pipe is unbuffered: both peers writing their version strings before + // reading deadlocks, so the successful-handshake case needs a real socket. + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer func() { _ = listener.Close() }() + + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + signer, err := ssh.NewSignerFromKey(key) + require.NoError(t, err) + serverConfig := &ssh.ServerConfig{NoClientAuth: true} + serverConfig.AddHostKey(signer) + go func() { + serverEnd, err := listener.Accept() + if err != nil { + return + } + _, chans, reqs, err := ssh.NewServerConn(serverEnd, serverConfig) + if err != nil { + return + } + go ssh.DiscardRequests(reqs) + for ch := range chans { + _ = ch.Reject(ssh.UnknownChannelType, "no channels in test") + } + }() + + clientEnd, err := net.Dial("tcp", listener.Addr().String()) + require.NoError(t, err) + conn := &deadlineRecorder{Conn: clientEnd} + client, err := clientFromConn(conn, "test", nil, 5*time.Second) + + require.NoError(t, err) + require.NotNil(t, client) + defer func() { _ = client.Close() }() + assert.Empty(t, conn.deadlines, "established session must stay free of deadlines") +} + +// slowReadConn adds one-way latency to reads, simulating a high-latency link +// where the peer is alive but every message arrives late. +type slowReadConn struct { + net.Conn + delay time.Duration +} + +func (c *slowReadConn) Read(b []byte) (int, error) { + time.Sleep(c.delay) + return c.Conn.Read(b) +} + +func TestClientFromConn_SlowPeerStillCompletes(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer func() { _ = listener.Close() }() + + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + signer, err := ssh.NewSignerFromKey(key) + require.NoError(t, err) + serverConfig := &ssh.ServerConfig{NoClientAuth: true} + serverConfig.AddHostKey(signer) + go func() { + serverEnd, err := listener.Accept() + if err != nil { + return + } + _, chans, reqs, err := ssh.NewServerConn(serverEnd, serverConfig) + if err != nil { + return + } + go ssh.DiscardRequests(reqs) + for ch := range chans { + _ = ch.Reject(ssh.UnknownChannelType, "no channels in test") + } + }() + + clientEnd, err := net.Dial("tcp", listener.Addr().String()) + require.NoError(t, err) + conn := &slowReadConn{Conn: clientEnd, delay: 30 * time.Millisecond} + + client, err := clientFromConn(conn, "test", nil, 500*time.Millisecond) + require.NoError(t, err, "slow-but-alive peer must not trip the idle timeout") + require.NotNil(t, client) + defer func() { _ = client.Close() }() +}