From 15bc325c67081c1876959816acda5767f0ef8a9e Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 25 May 2026 15:18:56 -0500 Subject: [PATCH 01/10] feat: connection-level keepalive and ConnectionClosingCallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hoists the keep-alive ticker loop out of session.handleRequests into a new connection-level goroutine in Server.HandleConn. It runs for the lifetime of the transport (using sshConn.SendRequest with the "keepalive@openssh.com" global request) rather than per-session, so a dead transport is detected even when no session is active — e.g., an idle ControlMaster after the outer ssh -O exit but before EOF propagates through the proxy chain. After ClientAliveCountMax consecutive intervals with no successful reply, sshConn is closed so HandleConn's channel loop unblocks. Also adds ConnectionClosingCallback, invoked synchronously the moment the inbound channels stream closes (before any defers, before sshConn.Wait()). Unlike ConnectionCompleteCallback this fires reliably even when the transport is permanently stuck. The callback receives the ssh.Context so per-connection state stashed via ctx.SetValue is reachable directly, removing the need for downstream callers to maintain a sync.Map keyed by *gossh.ServerConn. The per-session keepalive ticker driver in session.handleRequests is removed (the connection-level loop now drives the shared ctx.KeepAlive()); request-handler reset via KeepAliveRequestHandler still functions unchanged. --- server.go | 69 +++++++++++ server_test.go | 56 +++++++++ session.go | 316 ++++++++++++++++++++++--------------------------- ssh.go | 11 ++ 4 files changed, 276 insertions(+), 176 deletions(-) diff --git a/server.go b/server.go index 6f3d43f..dbedb4c 100644 --- a/server.go +++ b/server.go @@ -63,6 +63,7 @@ type Server struct { // succeed, never both. ConnectionFailedCallback ConnectionFailedCallback // callback to report connection failures ConnectionCompleteCallback ConnectionCompleteCallback // callback to report connection completion + ConnectionClosingCallback ConnectionClosingCallback // callback invoked synchronously when the transport begins closing, before sshConn.Wait() IdleTimeout time.Duration // connection timeout when no activity, none if empty MaxTimeout time.Duration // absolute connection timeout, none if empty @@ -323,6 +324,18 @@ func (srv *Server) HandleConn(newConn net.Conn) { applyConnMetadata(ctx, sshConn) // To prevent race conditions, we need to configure the keep-alive before goroutines kick off applyKeepAlive(ctx, srv.ClientAliveInterval, srv.ClientAliveCountMax) + + // Connection-level keep-alive: runs for the lifetime of the transport, + // independent of whether any session is active. This is what detects a + // dead transport between sessions (e.g., an idle ControlMaster whose + // outer ssh has been -O exit'd but whose ProxyCommand chain hasn't + // propagated EOF). On timeout we close sshConn so HandleConn's + // `for ch := range chans` loop unblocks and the closing/complete + // callbacks can fire. + keepAliveDone := make(chan struct{}) + go srv.connectionKeepAlive(ctx, sshConn, keepAliveDone) + defer close(keepAliveDone) + // go gossh.DiscardRequests(reqs) go srv.handleRequests(ctx, reqs) for ch := range chans { @@ -336,6 +349,62 @@ func (srv *Server) HandleConn(newConn net.Conn) { } go handler(srv, sshConn, ch, ctx) } + + // Fire the closing callback synchronously, before any defers (including + // ConnectionCompleteCallback's sshConn.Wait()) run. This is the only + // hook downstream callers can rely on when the transport is stuck — + // Wait() may never return. + if srv.ConnectionClosingCallback != nil { + srv.ConnectionClosingCallback(ctx, sshConn) + } +} + +// connectionKeepAlive drives transport-level keep-alive pings for the life +// of sshConn. It uses the gossh global-request channel +// (sshConn.SendRequest), which works whether or not any session channels +// exist. After ClientAliveCountMax consecutive intervals with no successful +// reply, sshConn is closed so HandleConn unblocks. Stops when `done` +// closes (HandleConn returning). +func (srv *Server) connectionKeepAlive( + ctx Context, + sshConn *gossh.ServerConn, + done <-chan struct{}, +) { + interval := srv.ClientAliveInterval + countMax := srv.ClientAliveCountMax + if interval <= 0 || countMax <= 0 { + return + } + + // Reuse the SessionKeepAlive already stashed on ctx so request-handler + // resets (KeepAliveRequestHandler) and metrics keep working. + keepAlive := ctx.KeepAlive() + + inFlight := make(chan struct{}, 1) + for { + select { + case <-done: + return + case <-keepAlive.Ticks(): + if keepAlive.TimeIsUp() { + _ = sshConn.Close() + return + } + select { + case inFlight <- struct{}{}: + default: + continue + } + go func() { + defer func() { <-inFlight }() + _, _, err := sshConn.SendRequest(keepAliveRequestType, true, nil) + keepAlive.ServerRequestedKeepAliveCallback() + if err == nil { + keepAlive.Reset() + } + }() + } + } } func (srv *Server) handleRequests(ctx Context, in <-chan *gossh.Request) { diff --git a/server_test.go b/server_test.go index 3cc0ab7..1613386 100644 --- a/server_test.go +++ b/server_test.go @@ -4,8 +4,11 @@ import ( "bytes" "context" "io" + "sync/atomic" "testing" "time" + + gossh "golang.org/x/crypto/ssh" ) func TestAddHostKey(t *testing.T) { @@ -80,6 +83,59 @@ func TestServerShutdown(t *testing.T) { } } +// TestConnectionClosingCallback verifies the closing callback fires +// synchronously when HandleConn observes the channels stream close, even +// when the client disconnects abruptly. It must fire before +// ConnectionCompleteCallback (which blocks on sshConn.Wait()). +func TestConnectionClosingCallback(t *testing.T) { + t.Parallel() + + closingFired := make(chan struct{}) + completeFired := make(chan struct{}) + var closingCtx atomic.Value // ssh.Context observed at callback time + + srv := &Server{ + Handler: func(s Session) { + _, _ = io.WriteString(s, "hi") + }, + ConnectionClosingCallback: func(ctx Context, _ *gossh.ServerConn) { + closingCtx.Store(ctx) + close(closingFired) + }, + ConnectionCompleteCallback: func(_ *gossh.ServerConn, _ error) { + close(completeFired) + }, + } + + l := newLocalTCPListener() + go func() { _ = srv.serveOnce(l) }() + + sess, client, _ := newClientSession(t, l.Addr().String(), nil) + if err := sess.Run(""); err != nil && err != io.EOF { + t.Fatalf("session run: %v", err) + } + _ = sess.Close() + _ = client.Close() + + select { + case <-closingFired: + case <-time.After(2 * time.Second): + t.Fatal("ConnectionClosingCallback did not fire within 2s of client disconnect") + } + + if got := closingCtx.Load(); got == nil { + t.Fatal("ConnectionClosingCallback received nil Context") + } + + // Complete callback should also fire after closing (Wait returns once + // the underlying TCP closes); it should not race ahead of closing. + select { + case <-completeFired: + case <-time.After(2 * time.Second): + t.Fatal("ConnectionCompleteCallback did not fire") + } +} + func TestServerClose(t *testing.T) { l := newLocalTCPListener() s := &Server{ diff --git a/session.go b/session.go index f21988b..643a5fd 100644 --- a/session.go +++ b/session.go @@ -4,8 +4,6 @@ import ( "bytes" "errors" "fmt" - "io" - "log" "net" "sync" @@ -282,203 +280,169 @@ func (sess *session) Break(c chan<- bool) { sess.breakCh = c } -func (sess *session) handleRequests(ctx Context, reqs <-chan *gossh.Request) { - keepAlive := ctx.KeepAlive() - defer keepAlive.Close() - - var keepAliveRequestInProgress sync.Mutex - for { - select { - case <-keepAlive.Ticks(): - if keepAlive.TimeIsUp() { - log.Println("Keep-alive reply not received. Close down the session.") - _ = sess.Close() - return +func (sess *session) handleRequests(_ Context, reqs <-chan *gossh.Request) { + // Transport-level keep-alive is driven from Server.HandleConn so it + // runs for the full connection lifetime, not just while a session is + // active. We only need to route per-session channel requests here. + defer func() { + // winch is created on pty-req; close it once when this loop + // exits (i.e., reqs closed) so any consumer ranging over it + // terminates cleanly. + if sess.winch != nil { + close(sess.winch) + } + }() + for req := range reqs { + switch req.Type { + case "shell", "exec": + if sess.handled { + _ = req.Reply(false, nil) + continue } - done := keepAliveRequestInProgress.TryLock() - if !done { + payload := struct{ Value string }{} + _ = gossh.Unmarshal(req.Payload, &payload) + sess.rawCmd = payload.Value + + // If there's a session policy callback, we need to confirm before + // accepting the session. + if sess.sessReqCb != nil && !sess.sessReqCb(sess, req.Type) { + sess.rawCmd = "" + _ = req.Reply(false, nil) continue } + sess.handled = true + _ = req.Reply(true, nil) + go func() { - defer keepAliveRequestInProgress.Unlock() - - // Server-initiated keep-alive flow on the client side: - // client: receive packet: type 98 (SSH_MSG_CHANNEL_REQUEST) - // client: client_input_channel_req: channel 0 rtype keepalive@openssh.com reply 1 - // client: send packet: type 100 (SSH_MSG_CHANNEL_FAILURE) - // - // Apparently, OpenSSH client always replies with 100, but it does not matter - // as the server considers it as alive (only the response status is ignored). - _, err := sess.SendRequest(keepAliveRequestType, true, nil) - keepAlive.ServerRequestedKeepAliveCallback() - if err != nil && err != io.EOF { - log.Printf("Sending keep-alive request failed: %v", err) - } else if err == nil { - keepAlive.Reset() - } + sess.handler(sess) + _ = sess.Exit(0) }() - case req, ok := <-reqs: - if !ok { - return + case "subsystem": + if sess.handled { + _ = req.Reply(false, nil) + continue } - switch req.Type { - case "shell", "exec": - if sess.handled { - _ = req.Reply(false, nil) - continue - } + payload := struct{ Value string }{} + _ = gossh.Unmarshal(req.Payload, &payload) + sess.subsystem = payload.Value - payload := struct{ Value string }{} - _ = gossh.Unmarshal(req.Payload, &payload) - sess.rawCmd = payload.Value - - // If there's a session policy callback, we need to confirm before - // accepting the session. - if sess.sessReqCb != nil && !sess.sessReqCb(sess, req.Type) { - sess.rawCmd = "" - _ = req.Reply(false, nil) - continue - } - - sess.handled = true - _ = req.Reply(true, nil) - - go func() { - sess.handler(sess) - _ = sess.Exit(0) - }() - case "subsystem": - if sess.handled { - _ = req.Reply(false, nil) - continue - } - - payload := struct{ Value string }{} - _ = gossh.Unmarshal(req.Payload, &payload) - sess.subsystem = payload.Value - - // If there's a session policy callback, we need to confirm before - // accepting the session. - if sess.sessReqCb != nil && !sess.sessReqCb(sess, req.Type) { - sess.rawCmd = "" - _ = req.Reply(false, nil) - continue - } + // If there's a session policy callback, we need to confirm before + // accepting the session. + if sess.sessReqCb != nil && !sess.sessReqCb(sess, req.Type) { + sess.rawCmd = "" + _ = req.Reply(false, nil) + continue + } - handler := sess.subsystemHandlers[payload.Value] - if handler == nil { - handler = sess.subsystemHandlers["default"] - } - if handler == nil { - _ = req.Reply(false, nil) - continue - } + handler := sess.subsystemHandlers[payload.Value] + if handler == nil { + handler = sess.subsystemHandlers["default"] + } + if handler == nil { + _ = req.Reply(false, nil) + continue + } - sess.handled = true - _ = req.Reply(true, nil) + sess.handled = true + _ = req.Reply(true, nil) - go func() { - handler(sess) - _ = sess.Exit(0) - }() - case "env": - if sess.handled { - _ = req.Reply(false, nil) - continue - } - var kv struct{ Key, Value string } - _ = gossh.Unmarshal(req.Payload, &kv) - sess.env = append(sess.env, fmt.Sprintf("%s=%s", kv.Key, kv.Value)) - _ = req.Reply(true, nil) - case "signal": - var payload struct{ Signal string } - _ = gossh.Unmarshal(req.Payload, &payload) - sess.sigMu.Lock() - if sess.sigCh != nil { - sess.sigCh <- Signal(payload.Signal) - } else if len(sess.sigBuf) < maxSigBufSize { - sess.sigBuf = append(sess.sigBuf, Signal(payload.Signal)) - } - sess.sigMu.Unlock() - case "pty-req": - if sess.handled || sess.pty != nil { - _ = req.Reply(false, nil) - continue - } - ptyReq, ok := parsePtyRequest(req.Payload) + go func() { + handler(sess) + _ = sess.Exit(0) + }() + case "env": + if sess.handled { + _ = req.Reply(false, nil) + continue + } + var kv struct{ Key, Value string } + _ = gossh.Unmarshal(req.Payload, &kv) + sess.env = append(sess.env, fmt.Sprintf("%s=%s", kv.Key, kv.Value)) + _ = req.Reply(true, nil) + case "signal": + var payload struct{ Signal string } + _ = gossh.Unmarshal(req.Payload, &payload) + sess.sigMu.Lock() + if sess.sigCh != nil { + sess.sigCh <- Signal(payload.Signal) + } else if len(sess.sigBuf) < maxSigBufSize { + sess.sigBuf = append(sess.sigBuf, Signal(payload.Signal)) + } + sess.sigMu.Unlock() + case "pty-req": + if sess.handled || sess.pty != nil { + _ = req.Reply(false, nil) + continue + } + ptyReq, ok := parsePtyRequest(req.Payload) + if !ok { + _ = req.Reply(false, nil) + continue + } + if sess.ptyCb != nil { + ok := sess.ptyCb(sess.ctx, ptyReq) if !ok { _ = req.Reply(false, nil) continue } - if sess.ptyCb != nil { - ok := sess.ptyCb(sess.ctx, ptyReq) - if !ok { - _ = req.Reply(false, nil) - continue - } - } - sess.pty = &ptyReq - sess.winch = make(chan Window, 1) - sess.winch <- ptyReq.Window - defer func() { - // when reqs is closed - close(sess.winch) - }() - _ = req.Reply(ok, nil) - case x11RequestType: - if sess.handled || sess.x11 != nil { - _ = req.Reply(false, nil) - continue - } - x11Req, ok := parseX11Request(req.Payload) + } + sess.pty = &ptyReq + sess.winch = make(chan Window, 1) + sess.winch <- ptyReq.Window + _ = req.Reply(ok, nil) + case x11RequestType: + if sess.handled || sess.x11 != nil { + _ = req.Reply(false, nil) + continue + } + x11Req, ok := parseX11Request(req.Payload) + if !ok { + _ = req.Reply(false, nil) + continue + } + if sess.x11Cb != nil { + ok := sess.x11Cb(sess.ctx, x11Req) if !ok { _ = req.Reply(false, nil) continue } - if sess.x11Cb != nil { - ok := sess.x11Cb(sess.ctx, x11Req) - if !ok { - _ = req.Reply(false, nil) - continue - } - } - sess.x11 = &x11Req - _ = req.Reply(ok, nil) - case "window-change": - if sess.pty == nil { - _ = req.Reply(false, nil) - continue - } - win, _, ok := parseWindow(req.Payload) - if ok { - sess.pty.Window = win - sess.winch <- win - } - _ = req.Reply(ok, nil) - case agentRequestType: - // TODO: option/callback to allow agent forwarding - SetAgentRequested(sess.ctx) - _ = req.Reply(true, nil) - case keepAliveRequestType: - if req.WantReply { - _ = req.Reply(true, nil) - } - case "break": - ok := false - sess.Lock() - if sess.breakCh != nil { - sess.breakCh <- true - ok = true - } - _ = req.Reply(ok, nil) - sess.Unlock() - default: - // TODO: debug log + } + sess.x11 = &x11Req + _ = req.Reply(ok, nil) + case "window-change": + if sess.pty == nil { _ = req.Reply(false, nil) + continue + } + win, _, ok := parseWindow(req.Payload) + if ok { + sess.pty.Window = win + sess.winch <- win + } + _ = req.Reply(ok, nil) + case agentRequestType: + // TODO: option/callback to allow agent forwarding + SetAgentRequested(sess.ctx) + _ = req.Reply(true, nil) + case keepAliveRequestType: + if req.WantReply { + _ = req.Reply(true, nil) + } + case "break": + ok := false + sess.Lock() + if sess.breakCh != nil { + sess.breakCh <- true + ok = true } + _ = req.Reply(ok, nil) + sess.Unlock() + default: + // TODO: debug log + _ = req.Reply(false, nil) } } } diff --git a/ssh.go b/ssh.go index 2bbfc10..39708d1 100644 --- a/ssh.go +++ b/ssh.go @@ -91,6 +91,17 @@ type ConnectionFailedCallback func(conn net.Conn, err error) // Please note: the ServerConn is closed at this point. type ConnectionCompleteCallback func(conn *gossh.ServerConn, err error) +// ConnectionClosingCallback is invoked synchronously the moment HandleConn +// observes the inbound channels stream close (i.e., the SSH transport is +// ending), BEFORE sshConn.Wait() is called and BEFORE any deferred cleanup +// runs. Unlike ConnectionCompleteCallback this does not block on Wait(), so +// it fires reliably even when the underlying transport is permanently stuck +// (e.g., a stalled ProxyCommand chain). Use it for fast-path resource +// cleanup. The Context is the same one threaded through auth and channel +// handlers, so per-connection state stashed via ctx.SetValue is reachable +// without an external sync.Map keyed by *gossh.ServerConn. +type ConnectionClosingCallback func(ctx Context, conn *gossh.ServerConn) + // Window represents the size of a PTY window. // // See https://datatracker.ietf.org/doc/html/rfc4254#section-6.2 From 46dc9ad1f00f6b693e299847e187da0e710f8dab Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 25 May 2026 15:26:06 -0500 Subject: [PATCH 02/10] fix(keepalive): plug ticker leak, bound SendRequest, restore observability MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses review feedback on the connection-level keep-alive change: - Add `defer keepAlive.Close()` in connectionKeepAlive so the underlying time.Ticker is stopped deterministically when the goroutine exits (previously leaked until GC). - Race sshConn.SendRequest against time.After(interval) so a stalled transport can't pin the inner goroutine forever; the buffered replyCh ensures the inner goroutine never blocks on send after the parent moves on. - Restore log output that the per-session driver previously emitted: failed SendRequest, per-request timeout, the keep-alive timeout-close decision, and sshConn.Close errors. - Tighten ConnectionClosingCallback godoc: it only fires reliably on a stuck transport when ClientAliveInterval is set. Without keep-alive there is nothing to break the gossh channel range loop. - Replace the field-level comment that duplicated the type doc with a pointer, and drop the HandleConn inline comment that named another callback's internals (rot risk). Tests: - TestConnectionClosingCallback now uses a single ordered event channel to assert closing fires strictly before complete. - New TestConnectionKeepAliveClosesStalledConn dials with raw gossh.NewClientConn and drops incoming requests without replying, asserting ConnectionClosingCallback fires after roughly ClientAliveInterval * ClientAliveCountMax. This is the regression test for the original bug — without the connection-level keep-alive it would hang. --- server.go | 38 +++++++++++++----- server_test.go | 103 ++++++++++++++++++++++++++++++++++++++++++------- ssh.go | 9 +++-- 3 files changed, 124 insertions(+), 26 deletions(-) diff --git a/server.go b/server.go index dbedb4c..470cdb4 100644 --- a/server.go +++ b/server.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "log" "net" "sync" "time" @@ -63,7 +64,7 @@ type Server struct { // succeed, never both. ConnectionFailedCallback ConnectionFailedCallback // callback to report connection failures ConnectionCompleteCallback ConnectionCompleteCallback // callback to report connection completion - ConnectionClosingCallback ConnectionClosingCallback // callback invoked synchronously when the transport begins closing, before sshConn.Wait() + ConnectionClosingCallback ConnectionClosingCallback // see ConnectionClosingCallback type doc IdleTimeout time.Duration // connection timeout when no activity, none if empty MaxTimeout time.Duration // absolute connection timeout, none if empty @@ -350,10 +351,11 @@ func (srv *Server) HandleConn(newConn net.Conn) { go handler(srv, sshConn, ch, ctx) } - // Fire the closing callback synchronously, before any defers (including - // ConnectionCompleteCallback's sshConn.Wait()) run. This is the only - // hook downstream callers can rely on when the transport is stuck — - // Wait() may never return. + // Fire the closing callback synchronously, before any deferred cleanup + // runs, so downstream callers have a deterministic hook the moment the + // channels stream ends. Deferred hooks that block on the transport + // (e.g., waiting on the mux) may be delayed indefinitely on a stuck + // transport; this path is not. if srv.ConnectionClosingCallback != nil { srv.ConnectionClosingCallback(ctx, sshConn) } @@ -379,6 +381,7 @@ func (srv *Server) connectionKeepAlive( // Reuse the SessionKeepAlive already stashed on ctx so request-handler // resets (KeepAliveRequestHandler) and metrics keep working. keepAlive := ctx.KeepAlive() + defer keepAlive.Close() inFlight := make(chan struct{}, 1) for { @@ -387,7 +390,13 @@ func (srv *Server) connectionKeepAlive( return case <-keepAlive.Ticks(): if keepAlive.TimeIsUp() { - _ = sshConn.Close() + log.Printf( + "ssh: connection keep-alive timeout after %d intervals; closing transport", + countMax, + ) + if err := sshConn.Close(); err != nil { + log.Printf("ssh: failed to close stalled transport: %v", err) + } return } select { @@ -397,10 +406,21 @@ func (srv *Server) connectionKeepAlive( } go func() { defer func() { <-inFlight }() - _, _, err := sshConn.SendRequest(keepAliveRequestType, true, nil) + replyCh := make(chan error, 1) + go func() { + _, _, err := sshConn.SendRequest(keepAliveRequestType, true, nil) + replyCh <- err + }() keepAlive.ServerRequestedKeepAliveCallback() - if err == nil { - keepAlive.Reset() + select { + case err := <-replyCh: + if err == nil { + keepAlive.Reset() + } else { + log.Printf("ssh: keepalive request failed: %v", err) + } + case <-time.After(interval): + log.Printf("ssh: keepalive request timed out after %s", interval) } }() } diff --git a/server_test.go b/server_test.go index 1613386..bf6200d 100644 --- a/server_test.go +++ b/server_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "io" + "net" "sync/atomic" "testing" "time" @@ -90,8 +91,10 @@ func TestServerShutdown(t *testing.T) { func TestConnectionClosingCallback(t *testing.T) { t.Parallel() - closingFired := make(chan struct{}) - completeFired := make(chan struct{}) + // Use a single ordered channel so a regression that swapped the order of + // the two callbacks would be caught. Reading from independent channels + // with separate timeouts would not detect such a swap. + events := make(chan string, 2) var closingCtx atomic.Value // ssh.Context observed at callback time srv := &Server{ @@ -100,10 +103,10 @@ func TestConnectionClosingCallback(t *testing.T) { }, ConnectionClosingCallback: func(ctx Context, _ *gossh.ServerConn) { closingCtx.Store(ctx) - close(closingFired) + events <- "closing" }, ConnectionCompleteCallback: func(_ *gossh.ServerConn, _ error) { - close(completeFired) + events <- "complete" }, } @@ -117,22 +120,96 @@ func TestConnectionClosingCallback(t *testing.T) { _ = sess.Close() _ = client.Close() - select { - case <-closingFired: - case <-time.After(2 * time.Second): - t.Fatal("ConnectionClosingCallback did not fire within 2s of client disconnect") + readEvent := func() string { + select { + case ev := <-events: + return ev + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for callback event") + return "" + } + } + + if first := readEvent(); first != "closing" { + t.Fatalf("expected first event to be \"closing\", got %q", first) + } + if second := readEvent(); second != "complete" { + t.Fatalf("expected second event to be \"complete\", got %q", second) } if got := closingCtx.Load(); got == nil { t.Fatal("ConnectionClosingCallback received nil Context") } +} + +// TestConnectionKeepAliveClosesStalledConn is the regression test for the +// connection-level keep-alive: when a client stops responding to keepalive +// global requests (e.g., idle ControlMaster whose peer is wedged) and no +// session is open, the server's connectionKeepAlive goroutine must close +// sshConn after roughly ClientAliveInterval * ClientAliveCountMax, which +// unblocks HandleConn's `for ch := range chans` loop and fires +// ConnectionClosingCallback. Before the fix, the keep-alive was only driven +// from within an active session, so an idle-but-stuck transport would +// linger forever. +// +// We dial with a raw gossh.NewClientConn and intentionally do NOT reply to +// the incoming requests channel, simulating a stuck peer. We do not open a +// session — the bug is specifically about the no-session case. +func TestConnectionKeepAliveClosesStalledConn(t *testing.T) { + t.Parallel() + + closingFired := make(chan struct{}) + + srv := &Server{ + Handler: func(_ Session) {}, + ClientAliveInterval: 100 * time.Millisecond, + ClientAliveCountMax: 2, + ConnectionClosingCallback: func(_ Context, _ *gossh.ServerConn) { + close(closingFired) + }, + } - // Complete callback should also fire after closing (Wait returns once - // the underlying TCP closes); it should not race ahead of closing. + l := newLocalTCPListener() + defer func() { _ = l.Close() }() + go func() { _ = srv.serveOnce(l) }() + + cfg := &gossh.ClientConfig{ + User: "testuser", + Auth: []gossh.AuthMethod{gossh.Password("testpass")}, + HostKeyCallback: gossh.InsecureIgnoreHostKey(), //nolint:gosec // test code + } + netConn, err := net.Dial("tcp", l.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + sshConn, chans, reqs, err := gossh.NewClientConn(netConn, l.Addr().String(), cfg) + if err != nil { + t.Fatalf("NewClientConn: %v", err) + } + defer func() { _ = sshConn.Close() }() + + // Drain both streams WITHOUT replying. By not calling req.Reply on + // incoming global requests (including keepalive@openssh.com), we + // simulate a stuck peer. The server's SendRequest with wantReply=true + // will not see a response and connectionKeepAlive will eventually + // close sshConn after ClientAliveCountMax intervals. + go func() { + for range chans { //nolint:revive // intentional drain + } + }() + go func() { + for req := range reqs { + _ = req // intentionally never reply + } + }() + + // 100ms * 2 = 200ms expected; allow generous slack for CI scheduling + // and any per-SendRequest timeout the other agent may layer on. select { - case <-completeFired: - case <-time.After(2 * time.Second): - t.Fatal("ConnectionCompleteCallback did not fire") + case <-closingFired: + case <-time.After(5 * time.Second): + t.Fatal("ConnectionClosingCallback did not fire; " + + "stalled client was not torn down by connection-level keep-alive") } } diff --git a/ssh.go b/ssh.go index 39708d1..ed1e604 100644 --- a/ssh.go +++ b/ssh.go @@ -94,10 +94,11 @@ type ConnectionCompleteCallback func(conn *gossh.ServerConn, err error) // ConnectionClosingCallback is invoked synchronously the moment HandleConn // observes the inbound channels stream close (i.e., the SSH transport is // ending), BEFORE sshConn.Wait() is called and BEFORE any deferred cleanup -// runs. Unlike ConnectionCompleteCallback this does not block on Wait(), so -// it fires reliably even when the underlying transport is permanently stuck -// (e.g., a stalled ProxyCommand chain). Use it for fast-path resource -// cleanup. The Context is the same one threaded through auth and channel +// runs. Use it for fast-path resource cleanup. Note: this fires when +// HandleConn observes the channels stream close. If the transport is +// permanently stuck and ClientAliveInterval is unset, neither this callback +// nor ConnectionCompleteCallback will fire — configure keep-alive to bound +// the wait. The Context is the same one threaded through auth and channel // handlers, so per-connection state stashed via ctx.SetValue is reachable // without an external sync.Map keyed by *gossh.ServerConn. type ConnectionClosingCallback func(ctx Context, conn *gossh.ServerConn) From 49e800dfe1f4ba53500ec235052c9fa9653bc88f Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 25 May 2026 15:35:38 -0500 Subject: [PATCH 03/10] feat(keepalive): match OpenSSH client_alive_check channel/global selection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OpenSSH's sshd sends keepalive@openssh.com as a per-channel request when at least one channel is open and as a global request otherwise (see serverloop.c client_alive_check). The previous connection-level implementation always sent a global request, which works with every real client (OpenSSH, gossh, libssh, paramiko all handle both forms) but diverged from sshd's wire behavior. This change tracks open channels per connection via a small openChannelSet stashed on the ssh.Context. Channel registration is transparent: each gossh.NewChannel is wrapped before being passed to the user-supplied ChannelHandler so Accept() registers the resulting Channel without any handler changes. Channel removal is lazy — when a keepalive SendRequest on a registered channel fails, that channel is pruned and the call falls through to the global request path. gossh exposes no Channel-close notification, so prune-on-failure is the cleanest approach. The keepalive loop now: - picks an open channel and calls ch.SendRequest if one exists, - prunes and falls back to sshConn.SendRequest on failure or empty set, - still bounds the send with the existing replyCh / time.After race, - still drives ctx.KeepAlive() for metrics and Reset on success, - still closes sshConn on TimeIsUp. Test added: TestConnectionKeepAliveUsesChannelRequestWhenSessionOpen opens a raw gossh channel, observes channel-typed keepalives arriving on the per-channel request stream (and zero on the global stream) for ~1s, then closes the channel and observes global-typed keepalives arriving for ~1s. --- context.go | 4 +++ server.go | 92 ++++++++++++++++++++++++++++++++++++++++++++++---- server_test.go | 85 ++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 174 insertions(+), 7 deletions(-) diff --git a/context.go b/context.go index df9b3aa..60cc781 100644 --- a/context.go +++ b/context.go @@ -60,6 +60,10 @@ var ( // ContextKeyKeepAlive is a context key for use with Contexts in this package. // The associated value will be of type *SessionKeepAlive. ContextKeyKeepAlive = &contextKey{"keep-alive"} + + // ContextKeyOpenChannels is a context key for use with Contexts in this package. + // The associated value will be of type *openChannelSet. + ContextKeyOpenChannels = &contextKey{"open-channels"} ) // Context is a package specific context interface. It exposes connection diff --git a/server.go b/server.go index 470cdb4..4456b34 100644 --- a/server.go +++ b/server.go @@ -16,6 +16,60 @@ import ( // and ListenAndServeTLS methods after a call to Shutdown or Close. var ErrServerClosed = errors.New("ssh: Server closed") +// openChannelSet tracks accepted channels for a connection so that +// connectionKeepAlive can mirror OpenSSH's client_alive_check() behavior: +// when at least one channel is open, send the keepalive as a channel +// request on that channel; otherwise fall back to a global request. +type openChannelSet struct { + mu sync.Mutex + chans []gossh.Channel +} + +func (s *openChannelSet) add(c gossh.Channel) { + s.mu.Lock() + defer s.mu.Unlock() + s.chans = append(s.chans, c) +} + +func (s *openChannelSet) any() gossh.Channel { + s.mu.Lock() + defer s.mu.Unlock() + if len(s.chans) == 0 { + return nil + } + return s.chans[0] +} + +func (s *openChannelSet) remove(c gossh.Channel) { + s.mu.Lock() + defer s.mu.Unlock() + for i, ch := range s.chans { + if ch == c { + s.chans = append(s.chans[:i], s.chans[i+1:]...) + return + } + } +} + +// trackingNewChannel wraps a gossh.NewChannel so that successful Accept +// calls register the underlying channel with an openChannelSet. The +// ChannelHandler API receives a gossh.NewChannel and typically calls +// Accept() inside the handler, so wrapping at HandleConn dispatch time is +// the only place to observe acceptance for arbitrary external handlers +// without modifying them. +type trackingNewChannel struct { + gossh.NewChannel + onAccept func(gossh.Channel) +} + +func (t *trackingNewChannel) Accept() (gossh.Channel, <-chan *gossh.Request, error) { + ch, reqs, err := t.NewChannel.Accept() + if err == nil && t.onAccept != nil { + t.onAccept(ch) + } + return ch, reqs, err +} + // SubsystemHandler is a callback for handling SSH subsystem requests. type SubsystemHandler func(s Session) @@ -325,6 +379,8 @@ func (srv *Server) HandleConn(newConn net.Conn) { applyConnMetadata(ctx, sshConn) // To prevent race conditions, we need to configure the keep-alive before goroutines kick off applyKeepAlive(ctx, srv.ClientAliveInterval, srv.ClientAliveCountMax) + openChans := &openChannelSet{} + ctx.SetValue(ContextKeyOpenChannels, openChans) // Connection-level keep-alive: runs for the lifetime of the transport, // independent of whether any session is active. This is what detects a @@ -348,7 +404,8 @@ func (srv *Server) HandleConn(newConn net.Conn) { _ = ch.Reject(gossh.UnknownChannelType, "unsupported channel type") continue } - go handler(srv, sshConn, ch, ctx) + tracked := &trackingNewChannel{NewChannel: ch, onAccept: openChans.add} + go handler(srv, sshConn, tracked, ctx) } // Fire the closing callback synchronously, before any deferred cleanup @@ -362,11 +419,12 @@ func (srv *Server) HandleConn(newConn net.Conn) { } // connectionKeepAlive drives transport-level keep-alive pings for the life -// of sshConn. It uses the gossh global-request channel -// (sshConn.SendRequest), which works whether or not any session channels -// exist. After ClientAliveCountMax consecutive intervals with no successful -// reply, sshConn is closed so HandleConn unblocks. Stops when `done` -// closes (HandleConn returning). +// of sshConn. It mirrors OpenSSH's client_alive_check(): if at least one +// channel is open, the keepalive is sent as a SSH2_MSG_CHANNEL_REQUEST on +// that channel; otherwise it falls back to a SSH2_MSG_GLOBAL_REQUEST. After +// ClientAliveCountMax consecutive intervals with no successful reply, +// sshConn is closed so HandleConn unblocks. Stops when `done` closes +// (HandleConn returning). func (srv *Server) connectionKeepAlive( ctx Context, sshConn *gossh.ServerConn, @@ -383,6 +441,8 @@ func (srv *Server) connectionKeepAlive( keepAlive := ctx.KeepAlive() defer keepAlive.Close() + openChans, _ := ctx.Value(ContextKeyOpenChannels).(*openChannelSet) + inFlight := make(chan struct{}, 1) for { select { @@ -408,7 +468,25 @@ func (srv *Server) connectionKeepAlive( defer func() { <-inFlight }() replyCh := make(chan error, 1) go func() { - _, _, err := sshConn.SendRequest(keepAliveRequestType, true, nil) + // Mirror OpenSSH client_alive_check(): prefer a channel + // request on an open channel; fall back to a global + // request if no channel is open or the channel send + // fails (channel was closed mid-flight). + var err error + var ch gossh.Channel + if openChans != nil { + ch = openChans.any() + } + if ch != nil { + _, err = ch.SendRequest(keepAliveRequestType, true, nil) + if err != nil { + openChans.remove(ch) + ch = nil + } + } + if ch == nil { + _, _, err = sshConn.SendRequest(keepAliveRequestType, true, nil) + } replyCh <- err }() keepAlive.ServerRequestedKeepAliveCallback() diff --git a/server_test.go b/server_test.go index bf6200d..c44edc7 100644 --- a/server_test.go +++ b/server_test.go @@ -213,6 +213,91 @@ func TestConnectionKeepAliveClosesStalledConn(t *testing.T) { } } +// TestConnectionKeepAliveUsesChannelRequestWhenSessionOpen verifies that the +// server's connection-level keepalive mirrors OpenSSH's client_alive_check(): +// while at least one channel is open it sends keepalive@openssh.com as a +// channel request on that channel; after the channel closes it falls back +// to a global request. +func TestConnectionKeepAliveUsesChannelRequestWhenSessionOpen(t *testing.T) { + t.Parallel() + + srv := &Server{ + Handler: func(s Session) { + <-s.Context().Done() + }, + ClientAliveInterval: 100 * time.Millisecond, + ClientAliveCountMax: 100, // generous so the conn doesn't get torn down mid-test + } + + l := newLocalTCPListener() + defer func() { _ = l.Close() }() + go func() { _ = srv.serveOnce(l) }() + + cfg := &gossh.ClientConfig{ + User: "testuser", + Auth: []gossh.AuthMethod{gossh.Password("testpass")}, + HostKeyCallback: gossh.InsecureIgnoreHostKey(), //nolint:gosec // test code + } + netConn, err := net.Dial("tcp", l.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + sshConn, chans, globalReqs, err := gossh.NewClientConn(netConn, l.Addr().String(), cfg) + if err != nil { + t.Fatalf("NewClientConn: %v", err) + } + defer func() { _ = sshConn.Close() }() + go func() { + for range chans { //nolint:revive // intentional drain + } + }() + + var globalCount, channelCount atomic.Int64 + go func() { + for req := range globalReqs { + if req.Type == "keepalive@openssh.com" { + globalCount.Add(1) + } + _ = req.Reply(true, nil) + } + }() + + ch, chReqs, err := sshConn.OpenChannel("session", nil) + if err != nil { + t.Fatalf("OpenChannel: %v", err) + } + chDone := make(chan struct{}) + go func() { + defer close(chDone) + for req := range chReqs { + if req.Type == "keepalive@openssh.com" { + channelCount.Add(1) + } + if req.WantReply { + _ = req.Reply(true, nil) + } + } + }() + + // Observe ~10 intervals with the channel open. + time.Sleep(1 * time.Second) + if got := channelCount.Load(); got < 1 { + t.Fatalf("expected >=1 channel-typed keepalive while channel open, got %d", got) + } + if got := globalCount.Load(); got != 0 { + t.Fatalf("expected 0 global-typed keepalives while channel open, got %d", got) + } + + // Close the channel and observe the fallback to global requests. + _ = ch.Close() + <-chDone + before := globalCount.Load() + time.Sleep(1 * time.Second) + if got := globalCount.Load() - before; got < 1 { + t.Fatalf("expected >=1 global-typed keepalive after channel close, got %d", got) + } +} + func TestServerClose(t *testing.T) { l := newLocalTCPListener() s := &Server{ From b6cbe72895eb229caa116a61dc24deac6ec80129 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 25 May 2026 15:45:17 -0500 Subject: [PATCH 04/10] fix(keepalive): close OpenSSH-compliance gaps in channel-aware keepalive MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses five divergences from OpenSSH sshd surfaced during review. Channel-close hook (closes #1 of review) Previously openChannelSet.remove fired only on a SendRequest failure, so a connection that opened and cleanly closed many channels accumulated dead entries forever. any() always returned slot 0 — the oldest, most-likely-dead channel — degrading the OpenSSH-mirror mode into "always probe a dead channel, then fall back to global" within one tick. trackingNewChannel.Accept now proxies the per-channel request stream through a buffered (16, matching gossh's chanSize) forwarder goroutine; when the upstream reqs channel closes, an onClose callback prunes the set. Remove the time.After race (closes #2 of review) connectionKeepAlive used a replyCh + time.After(interval) race around SendRequest. A reply arriving at interval+epsilon left keepAlive.Reset() un-called, ticking the false-disconnect counter for live-but-slow clients. Removed the outer timeout entirely. The inFlight semaphore already prevents overlapping probes; TimeIsUp enforces the deadline at the next tick; sshConn.Close on TimeIsUp unblocks any hung SendRequest. Reset on any inbound traffic (closes #3 of review) OpenSSH resets its client-alive counter on every received packet. Added SessionKeepAlive.NotePeerActivity — like Reset but without bumping the KeepAliveReplyReceived metric. Called from Server.handleRequests (global requests), session.handleRequests (per-session requests), and the per-channel request forwarder in trackingNewChannel. Resets the ticker so the next probe fires ClientAliveInterval after the most recent activity, which also closes #4 (probe suppression on recent traffic) for free. Unwrap helper + ChannelHandlers doc (closes #5 of review) Added Unwrap() gossh.NewChannel on trackingNewChannel so downstream handlers that need the underlying type can recover it. Documented the wrapping on Server.ChannelHandlers. Test added: TestConnectionKeepAlivePrunesClosedChannels opens three channels, closes them, then asserts subsequent keepalives over a 1s window are global-only (0 channel-typed, >=1 global). Without the close hook, the dead channels in the set would force channel-typed probes that fail and prune one-at-a-time over many intervals. --- keepalive.go | 15 ++++++ server.go | 127 +++++++++++++++++++++++++++++++++++-------------- server_test.go | 94 ++++++++++++++++++++++++++++++++++++ session.go | 3 ++ 4 files changed, 202 insertions(+), 37 deletions(-) diff --git a/keepalive.go b/keepalive.go index 77a7321..e05aeaf 100644 --- a/keepalive.go +++ b/keepalive.go @@ -71,6 +71,21 @@ func (ska *SessionKeepAlive) Reset() { } } +// NotePeerActivity marks that traffic was received from the peer. Unlike +// Reset, it does not bump the KeepAliveReplyReceived metric — it's +// intended for arbitrary inbound packets, not specifically keepalive +// replies. The ticker is reset so the next probe fires +// clientAliveInterval after the most recent activity, matching OpenSSH's +// last_client_time logic. +func (ska *SessionKeepAlive) NotePeerActivity() { + ska.m.Lock() + defer ska.m.Unlock() + if ska.ticker != nil && !ska.closed { + ska.lastReceived = time.Now() + ska.ticker.Reset(ska.clientAliveInterval) + } +} + // Ticks returns the channel that fires on each keep-alive interval. func (ska *SessionKeepAlive) Ticks() <-chan time.Time { return ska.tickerCh diff --git a/server.go b/server.go index 4456b34..07af328 100644 --- a/server.go +++ b/server.go @@ -57,17 +57,56 @@ func (s *openChannelSet) remove(c gossh.Channel) { // Accept() inside the handler, so wrapping at HandleConn dispatch time is // the only place to observe acceptance for arbitrary external handlers // without modifying them. +// +// The per-channel request stream returned by Accept is also wrapped so +// that close of the underlying gossh stream (channel teardown) drops the +// channel from the openChannelSet, and so each inbound request bumps the +// keep-alive activity marker. type trackingNewChannel struct { gossh.NewChannel - onAccept func(gossh.Channel) + onAccept func(gossh.Channel) + onClose func(gossh.Channel) + notePeerActivity func() } +// trackingChanReqBuffer matches gossh's per-channel request channel +// buffer (see chanSize in golang.org/x/crypto/ssh/handshake.go). Keeping +// the size in sync avoids changing back-pressure semantics for handlers +// that did not previously block on a full request channel. +const trackingChanReqBuffer = 16 + func (t *trackingNewChannel) Accept() (gossh.Channel, <-chan *gossh.Request, error) { ch, reqs, err := t.NewChannel.Accept() - if err == nil && t.onAccept != nil { + if err != nil { + return ch, reqs, err + } + if t.onAccept != nil { t.onAccept(ch) } - return ch, reqs, err + wrapped := make(chan *gossh.Request, trackingChanReqBuffer) + go func() { + defer close(wrapped) + defer func() { + if t.onClose != nil { + t.onClose(ch) + } + }() + for r := range reqs { + if t.notePeerActivity != nil { + t.notePeerActivity() + } + wrapped <- r + } + }() + return ch, wrapped, nil +} + +// Unwrap returns the underlying gossh.NewChannel. Callers that received a +// NewChannel via a ChannelHandler and need access to the unwrapped value +// (e.g., for type assertions against a custom NewChannel implementation) +// can call this. +func (t *trackingNewChannel) Unwrap() gossh.NewChannel { + return t.NewChannel } // SubsystemHandler is a callback for handling SSH subsystem requests. @@ -126,6 +165,12 @@ type Server struct { // ChannelHandlers allow overriding the built-in session handlers or provide // extensions to the protocol, such as tcpip forwarding. By default only the // "session" handler is enabled. + // + // The gossh.NewChannel value passed to handlers may be wrapped by this + // package for keep-alive bookkeeping (tracking open channels and noting + // inbound activity). Handlers that need access to the unwrapped + // underlying value can type-assert to interface{ Unwrap() gossh.NewChannel } + // and call Unwrap. ChannelHandlers map[string]ChannelHandler // RequestHandlers allow overriding the server-level request handlers or @@ -404,7 +449,16 @@ func (srv *Server) HandleConn(newConn net.Conn) { _ = ch.Reject(gossh.UnknownChannelType, "unsupported channel type") continue } - tracked := &trackingNewChannel{NewChannel: ch, onAccept: openChans.add} + tracked := &trackingNewChannel{ + NewChannel: ch, + onAccept: openChans.add, + onClose: openChans.remove, + notePeerActivity: func() { + if ka := ctx.KeepAlive(); ka != nil { + ka.NotePeerActivity() + } + }, + } go handler(srv, sshConn, tracked, ctx) } @@ -430,9 +484,8 @@ func (srv *Server) connectionKeepAlive( sshConn *gossh.ServerConn, done <-chan struct{}, ) { - interval := srv.ClientAliveInterval countMax := srv.ClientAliveCountMax - if interval <= 0 || countMax <= 0 { + if srv.ClientAliveInterval <= 0 || countMax <= 0 { return } @@ -466,39 +519,36 @@ func (srv *Server) connectionKeepAlive( } go func() { defer func() { <-inFlight }() - replyCh := make(chan error, 1) - go func() { - // Mirror OpenSSH client_alive_check(): prefer a channel - // request on an open channel; fall back to a global - // request if no channel is open or the channel send - // fails (channel was closed mid-flight). - var err error - var ch gossh.Channel - if openChans != nil { - ch = openChans.any() - } - if ch != nil { - _, err = ch.SendRequest(keepAliveRequestType, true, nil) - if err != nil { - openChans.remove(ch) - ch = nil - } - } - if ch == nil { - _, _, err = sshConn.SendRequest(keepAliveRequestType, true, nil) - } - replyCh <- err - }() keepAlive.ServerRequestedKeepAliveCallback() - select { - case err := <-replyCh: - if err == nil { - keepAlive.Reset() - } else { - log.Printf("ssh: keepalive request failed: %v", err) + // Mirror OpenSSH client_alive_check(): prefer a channel + // request on an open channel; fall back to a global + // request if no channel is open or the channel send + // fails (channel was closed mid-flight). + // + // No outer timeout is needed here: the inFlight semaphore + // already prevents overlapping probes, TimeIsUp() at the + // next tick enforces the deadline, and if SendRequest + // hangs forever it will be unblocked when the TimeIsUp + // branch closes sshConn. + var err error + var ch gossh.Channel + if openChans != nil { + ch = openChans.any() + } + if ch != nil { + _, err = ch.SendRequest(keepAliveRequestType, true, nil) + if err != nil { + openChans.remove(ch) + ch = nil } - case <-time.After(interval): - log.Printf("ssh: keepalive request timed out after %s", interval) + } + if ch == nil { + _, _, err = sshConn.SendRequest(keepAliveRequestType, true, nil) + } + if err == nil { + keepAlive.Reset() + } else { + log.Printf("ssh: keepalive request failed: %v", err) } }() } @@ -507,6 +557,9 @@ func (srv *Server) connectionKeepAlive( func (srv *Server) handleRequests(ctx Context, in <-chan *gossh.Request) { for req := range in { + if ka := ctx.KeepAlive(); ka != nil { + ka.NotePeerActivity() + } handler := srv.RequestHandlers[req.Type] if handler == nil { handler = srv.RequestHandlers["default"] diff --git a/server_test.go b/server_test.go index c44edc7..d040927 100644 --- a/server_test.go +++ b/server_test.go @@ -298,6 +298,100 @@ func TestConnectionKeepAliveUsesChannelRequestWhenSessionOpen(t *testing.T) { } } +// TestConnectionKeepAlivePrunesClosedChannels verifies that once all +// channels are closed, the server's connection-level keepalive falls +// back to global requests rather than retaining stale channel handles +// from the openChannelSet. Regression test for the case where +// openChans.any() returned a dead channel because per-channel removal +// only happened on SendRequest failure. +func TestConnectionKeepAlivePrunesClosedChannels(t *testing.T) { + t.Parallel() + + srv := &Server{ + Handler: func(s Session) { + <-s.Context().Done() + }, + ClientAliveInterval: 100 * time.Millisecond, + ClientAliveCountMax: 100, // generous so the conn isn't torn down mid-test + } + + l := newLocalTCPListener() + defer func() { _ = l.Close() }() + go func() { _ = srv.serveOnce(l) }() + + cfg := &gossh.ClientConfig{ + User: "testuser", + Auth: []gossh.AuthMethod{gossh.Password("testpass")}, + HostKeyCallback: gossh.InsecureIgnoreHostKey(), //nolint:gosec // test code + } + netConn, err := net.Dial("tcp", l.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + sshConn, chans, globalReqs, err := gossh.NewClientConn(netConn, l.Addr().String(), cfg) + if err != nil { + t.Fatalf("NewClientConn: %v", err) + } + defer func() { _ = sshConn.Close() }() + go func() { + for range chans { //nolint:revive // intentional drain + } + }() + + var globalCount, channelCount atomic.Int64 + go func() { + for req := range globalReqs { + if req.Type == "keepalive@openssh.com" { + globalCount.Add(1) + } + _ = req.Reply(true, nil) + } + }() + + // Open three channels, drain their request streams (replying to + // keepalives so the channel doesn't get rejected), then close them. + var chReqWg [3]chan struct{} + for i := 0; i < 3; i++ { + ch, chReqs, err := sshConn.OpenChannel("session", nil) + if err != nil { + t.Fatalf("OpenChannel[%d]: %v", i, err) + } + done := make(chan struct{}) + chReqWg[i] = done + go func() { + defer close(done) + for req := range chReqs { + if req.WantReply { + _ = req.Reply(true, nil) + } + } + }() + _ = ch.Close() + } + for i := 0; i < 3; i++ { + <-chReqWg[i] + } + + // Allow the server's request-proxy goroutines to observe the channel + // closes and prune the openChannelSet. + time.Sleep(200 * time.Millisecond) + + beforeGlobal := globalCount.Load() + beforeChannel := channelCount.Load() + time.Sleep(1 * time.Second) + gotChannel := channelCount.Load() - beforeChannel + gotGlobal := globalCount.Load() - beforeGlobal + if gotChannel != 0 { + t.Fatalf( + "expected 0 channel-typed keepalives after all channels closed, got %d", + gotChannel, + ) + } + if gotGlobal < 1 { + t.Fatalf("expected >=1 global-typed keepalive after all channels closed, got %d", gotGlobal) + } +} + func TestServerClose(t *testing.T) { l := newLocalTCPListener() s := &Server{ diff --git a/session.go b/session.go index 643a5fd..138ba44 100644 --- a/session.go +++ b/session.go @@ -293,6 +293,9 @@ func (sess *session) handleRequests(_ Context, reqs <-chan *gossh.Request) { } }() for req := range reqs { + if ka := sess.ctx.KeepAlive(); ka != nil { + ka.NotePeerActivity() + } switch req.Type { case "shell", "exec": if sess.handled { From c4d6e82852d553e3bf32ae564a8da2edb39e7e1f Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 25 May 2026 15:54:45 -0500 Subject: [PATCH 05/10] fix(keepalive): forwarder cancellation, dead-peer detection, test rigor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Forwarder goroutine in trackingNewChannel.Accept now selects on ctx.Done(); on cancel it drains upstream reqs (replying false to WantReply) so the deferred onClose runs and gossh's sender doesn't leak. Previously a handler that abandoned wrapped would leak the forwarder forever once its 16-slot buffer filled. - NotePeerActivity no longer bumps lastReceived. It still resets the ticker (suppressing redundant probes when traffic flows), but the dead-peer deadline used by TimeIsUp is now cleared only by replies to our own probes. A chatty-but-wedged peer (send-side stuck, recv-side streaming) is no longer falsely considered alive — this matches OpenSSH's client_alive_check, which counts unanswered server-initiated probes, not arbitrary inbound traffic. - TestConnectionKeepAlivePrunesClosedChannels now actually exercises the close hook: interval bumped to 1s so we can poll the openChannelSet via context BEFORE the next keepalive tick. The previous timings allowed the old prune-on-failure path to pass the test without the hook. - Exported NewChannelUnwrapper named interface so Unwrap() is discoverable via go doc. ChannelHandlers godoc updated. --- keepalive.go | 17 ++++---- server.go | 53 ++++++++++++++++++++---- server_test.go | 107 +++++++++++++++++++++++++++++++++++++------------ 3 files changed, 138 insertions(+), 39 deletions(-) diff --git a/keepalive.go b/keepalive.go index e05aeaf..a6ade5b 100644 --- a/keepalive.go +++ b/keepalive.go @@ -71,17 +71,20 @@ func (ska *SessionKeepAlive) Reset() { } } -// NotePeerActivity marks that traffic was received from the peer. Unlike -// Reset, it does not bump the KeepAliveReplyReceived metric — it's -// intended for arbitrary inbound packets, not specifically keepalive -// replies. The ticker is reset so the next probe fires -// clientAliveInterval after the most recent activity, matching OpenSSH's -// last_client_time logic. +// NotePeerActivity records that we received some inbound traffic from +// the peer. Used to suppress a redundant keep-alive probe when the +// connection is actively chatty: the ticker is bumped so the next probe +// fires clientAliveInterval after the most recent activity. We +// deliberately do NOT update lastReceived — the dead-peer deadline used +// by TimeIsUp is cleared ONLY by replies to our own keep-alive probes. +// Otherwise a peer with a wedged send-side that's still streaming +// inbound junk would never be detected as dead, which matches OpenSSH's +// client_alive_check semantics (it counts unanswered server-initiated +// probes, not inbound traffic). func (ska *SessionKeepAlive) NotePeerActivity() { ska.m.Lock() defer ska.m.Unlock() if ska.ticker != nil && !ska.closed { - ska.lastReceived = time.Now() ska.ticker.Reset(ska.clientAliveInterval) } } diff --git a/server.go b/server.go index 07af328..3c0b818 100644 --- a/server.go +++ b/server.go @@ -64,9 +64,11 @@ func (s *openChannelSet) remove(c gossh.Channel) { // keep-alive activity marker. type trackingNewChannel struct { gossh.NewChannel + set *openChannelSet onAccept func(gossh.Channel) onClose func(gossh.Channel) notePeerActivity func() + ctx Context } // trackingChanReqBuffer matches gossh's per-channel request channel @@ -91,20 +93,55 @@ func (t *trackingNewChannel) Accept() (gossh.Channel, <-chan *gossh.Request, err t.onClose(ch) } }() + // ctxDone may be nil if no ctx was wired in (defensive: keep the + // goroutine cancellable on connection teardown without + // requiring every caller to plumb ctx). A nil channel blocks + // forever in select, which gives the same behavior as a plain + // unconditional send if ctx is absent. + var ctxDone <-chan struct{} + if t.ctx != nil { + ctxDone = t.ctx.Done() + } for r := range reqs { if t.notePeerActivity != nil { t.notePeerActivity() } - wrapped <- r + select { + case wrapped <- r: + case <-ctxDone: + // Handler stopped draining and the connection is + // going away. Negatively reply to any want-reply + // request we were holding, then drain upstream so + // gossh's per-channel request goroutine doesn't + // leak, replying false to any subsequent + // want-reply requests as we go. + if r.WantReply { + _ = r.Reply(false, nil) + } + for r2 := range reqs { + if r2.WantReply { + _ = r2.Reply(false, nil) + } + } + return + } } }() return ch, wrapped, nil } -// Unwrap returns the underlying gossh.NewChannel. Callers that received a -// NewChannel via a ChannelHandler and need access to the unwrapped value -// (e.g., for type assertions against a custom NewChannel implementation) -// can call this. +// NewChannelUnwrapper is implemented by NewChannel implementations that +// wrap another NewChannel for internal bookkeeping. Channel handlers +// that need access to the underlying gossh.NewChannel (e.g., to type- +// assert against a custom implementation) can call Unwrap to recover +// it. This is needed because the library wraps every incoming +// NewChannel to track per-channel keep-alive activity; the wrapper is +// otherwise transparent. +type NewChannelUnwrapper interface { + Unwrap() gossh.NewChannel +} + +// Unwrap returns the underlying gossh.NewChannel. See NewChannelUnwrapper. func (t *trackingNewChannel) Unwrap() gossh.NewChannel { return t.NewChannel } @@ -169,8 +206,8 @@ type Server struct { // The gossh.NewChannel value passed to handlers may be wrapped by this // package for keep-alive bookkeeping (tracking open channels and noting // inbound activity). Handlers that need access to the unwrapped - // underlying value can type-assert to interface{ Unwrap() gossh.NewChannel } - // and call Unwrap. + // underlying value can type-assert to NewChannelUnwrapper and call + // Unwrap. ChannelHandlers map[string]ChannelHandler // RequestHandlers allow overriding the server-level request handlers or @@ -451,6 +488,7 @@ func (srv *Server) HandleConn(newConn net.Conn) { } tracked := &trackingNewChannel{ NewChannel: ch, + set: openChans, onAccept: openChans.add, onClose: openChans.remove, notePeerActivity: func() { @@ -458,6 +496,7 @@ func (srv *Server) HandleConn(newConn net.Conn) { ka.NotePeerActivity() } }, + ctx: ctx, } go handler(srv, sshConn, tracked, ctx) } diff --git a/server_test.go b/server_test.go index d040927..793285a 100644 --- a/server_test.go +++ b/server_test.go @@ -298,20 +298,34 @@ func TestConnectionKeepAliveUsesChannelRequestWhenSessionOpen(t *testing.T) { } } -// TestConnectionKeepAlivePrunesClosedChannels verifies that once all -// channels are closed, the server's connection-level keepalive falls -// back to global requests rather than retaining stale channel handles -// from the openChannelSet. Regression test for the case where -// openChans.any() returned a dead channel because per-channel removal -// only happened on SendRequest failure. +// TestConnectionKeepAlivePrunesClosedChannels verifies that the +// per-channel close hook prunes channels from the openChannelSet as +// soon as the client closes them, BEFORE the next keepalive probe +// fires. This is a regression test for the case where openChans.any() +// returned a dead channel because per-channel removal only happened on +// SendRequest failure (i.e., one probe was wasted on a dead channel +// before the prune-on-failure path kicked in). +// +// Strategy: use a long ClientAliveInterval (1s) so we can observe the +// state of openChans in the window between channel close and the first +// post-close probe. We capture the openChannelSet from inside the +// session handler so the test can inspect it without exporting helpers. func TestConnectionKeepAlivePrunesClosedChannels(t *testing.T) { t.Parallel() + openChansCh := make(chan *openChannelSet, 1) srv := &Server{ Handler: func(s Session) { - <-s.Context().Done() + ctx := s.Context() + if oc, ok := ctx.Value(ContextKeyOpenChannels).(*openChannelSet); ok { + select { + case openChansCh <- oc: + default: + } + } + <-ctx.Done() }, - ClientAliveInterval: 100 * time.Millisecond, + ClientAliveInterval: 1 * time.Second, ClientAliveCountMax: 100, // generous so the conn isn't torn down mid-test } @@ -338,7 +352,7 @@ func TestConnectionKeepAlivePrunesClosedChannels(t *testing.T) { } }() - var globalCount, channelCount atomic.Int64 + var globalCount atomic.Int64 go func() { for req := range globalReqs { if req.Type == "keepalive@openssh.com" { @@ -348,9 +362,11 @@ func TestConnectionKeepAlivePrunesClosedChannels(t *testing.T) { } }() - // Open three channels, drain their request streams (replying to - // keepalives so the channel doesn't get rejected), then close them. + // Open three sessions so that the server handler runs and registers + // channels in openChans. Use a "shell" request so DefaultSessionHandler + // considers the session "handled" and our top-level Handler runs. var chReqWg [3]chan struct{} + channels := make([]gossh.Channel, 0, 3) for i := 0; i < 3; i++ { ch, chReqs, err := sshConn.OpenChannel("session", nil) if err != nil { @@ -366,29 +382,70 @@ func TestConnectionKeepAlivePrunesClosedChannels(t *testing.T) { } } }() + // Send a shell request to trigger the user Handler so it can + // publish openChans into openChansCh. + if _, err := ch.SendRequest("shell", true, nil); err != nil { + t.Fatalf("shell request[%d]: %v", i, err) + } + channels = append(channels, ch) + } + + // Grab the openChannelSet from the first session that ran. + var openChans *openChannelSet + select { + case openChans = <-openChansCh: + case <-time.After(2 * time.Second): + t.Fatalf("never received openChannelSet from handler") + } + + // Sanity: all three channels are registered. + openChans.mu.Lock() + regBefore := len(openChans.chans) + openChans.mu.Unlock() + if regBefore != 3 { + t.Fatalf("expected 3 channels registered before close, got %d", regBefore) + } + + // Close all channels. + for _, ch := range channels { _ = ch.Close() } for i := 0; i < 3; i++ { <-chReqWg[i] } - // Allow the server's request-proxy goroutines to observe the channel - // closes and prune the openChannelSet. - time.Sleep(200 * time.Millisecond) - - beforeGlobal := globalCount.Load() - beforeChannel := channelCount.Load() - time.Sleep(1 * time.Second) - gotChannel := channelCount.Load() - beforeChannel - gotGlobal := globalCount.Load() - beforeGlobal - if gotChannel != 0 { + // Poll for the close hook to drain. ClientAliveInterval is 1s, so we + // have ample time to observe the prune BEFORE the next probe fires. + // If the close hook works, len(openChans.chans) drops to 0 quickly + // (just goroutine scheduling). If the hook is broken, it stays at 3 + // until the next keepalive tick fires and SendRequest fails. + deadline := time.Now().Add(800 * time.Millisecond) + for time.Now().Before(deadline) { + openChans.mu.Lock() + n := len(openChans.chans) + openChans.mu.Unlock() + if n == 0 { + break + } + time.Sleep(10 * time.Millisecond) + } + openChans.mu.Lock() + regAfter := len(openChans.chans) + openChans.mu.Unlock() + if regAfter != 0 { t.Fatalf( - "expected 0 channel-typed keepalives after all channels closed, got %d", - gotChannel, + "openChannelSet still has %d entries 800ms after close; close hook did not run", + regAfter, ) } - if gotGlobal < 1 { - t.Fatalf("expected >=1 global-typed keepalive after all channels closed, got %d", gotGlobal) + + // And verify that the next keepalive probe (which fires ~1s after the + // last reset, i.e. shortly after this point) goes out as a global + // request — confirming behavior end-to-end. + before := globalCount.Load() + time.Sleep(1500 * time.Millisecond) + if got := globalCount.Load() - before; got < 1 { + t.Fatalf("expected >=1 global-typed keepalive after channels closed, got %d", got) } } From dd59fb01e95bd9a63d2f412e1a529c489415bfb0 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 25 May 2026 15:59:20 -0500 Subject: [PATCH 06/10] fix(keepalive): restore OpenSSH-faithful peer-activity semantics Round-2 review surfaced an internal contradiction: NotePeerActivity reset the ticker but did not bump lastReceived. A chatty inbound peer would keep deferring the probe (via ticker.Reset) so TimeIsUp never got consulted, defeating dead-peer detection. The doc comment also misrepresented OpenSSH semantics. OpenSSH's ssh_packet_read_poll_seqnr (packet.c) zeros keep_alive_timeouts on every successfully-received packet of any type, and serverloop.c defers the next probe on inbound traffic. NotePeerActivity now matches: bumps lastReceived AND resets the ticker on any inbound traffic. The Reset metric counter is still reserved for actual probe replies. Also: - Removed unreachable if t.ctx != nil guard in trackingNewChannel. Construction site (HandleConn) always passes a non-nil ctx, and the prior fallback comment was wrong (a nil select case deadlocks on full buffer, doesn't degrade). - Tightened TestConnectionKeepAlivePrunesClosedChannels: interval 1s -> 3s, poll deadline 800ms -> 500ms. Added invariant assertion that no keepalive tick fires inside the assertion window so the test can't silently pass via the prune-on-failure fallback. --- keepalive.go | 21 +++++++++++---------- server.go | 10 +--------- server_test.go | 35 ++++++++++++++++++++++++++--------- 3 files changed, 38 insertions(+), 28 deletions(-) diff --git a/keepalive.go b/keepalive.go index a6ade5b..a407484 100644 --- a/keepalive.go +++ b/keepalive.go @@ -71,20 +71,21 @@ func (ska *SessionKeepAlive) Reset() { } } -// NotePeerActivity records that we received some inbound traffic from -// the peer. Used to suppress a redundant keep-alive probe when the -// connection is actively chatty: the ticker is bumped so the next probe -// fires clientAliveInterval after the most recent activity. We -// deliberately do NOT update lastReceived — the dead-peer deadline used -// by TimeIsUp is cleared ONLY by replies to our own keep-alive probes. -// Otherwise a peer with a wedged send-side that's still streaming -// inbound junk would never be detected as dead, which matches OpenSSH's -// client_alive_check semantics (it counts unanswered server-initiated -// probes, not inbound traffic). +// NotePeerActivity records that inbound traffic was observed from the +// peer. Like Reset, it bumps lastReceived (which clears the dead-peer +// deadline used by TimeIsUp) and resets the ticker (suppressing a +// redundant probe). Unlike Reset, it does NOT bump the +// KeepAliveReplyReceived metric — that's reserved for actual probe +// replies. This matches OpenSSH sshd: any successfully-received packet +// zeros keep_alive_timeouts (packet.c ssh_packet_read_poll_seqnr) and +// defers the next probe (serverloop.c). The unanswered-probe counter +// is a function of silence on the wire, not of probe-reply tracking +// specifically. func (ska *SessionKeepAlive) NotePeerActivity() { ska.m.Lock() defer ska.m.Unlock() if ska.ticker != nil && !ska.closed { + ska.lastReceived = time.Now() ska.ticker.Reset(ska.clientAliveInterval) } } diff --git a/server.go b/server.go index 3c0b818..f6f355a 100644 --- a/server.go +++ b/server.go @@ -93,15 +93,7 @@ func (t *trackingNewChannel) Accept() (gossh.Channel, <-chan *gossh.Request, err t.onClose(ch) } }() - // ctxDone may be nil if no ctx was wired in (defensive: keep the - // goroutine cancellable on connection teardown without - // requiring every caller to plumb ctx). A nil channel blocks - // forever in select, which gives the same behavior as a plain - // unconditional send if ctx is absent. - var ctxDone <-chan struct{} - if t.ctx != nil { - ctxDone = t.ctx.Done() - } + ctxDone := t.ctx.Done() for r := range reqs { if t.notePeerActivity != nil { t.notePeerActivity() diff --git a/server_test.go b/server_test.go index 793285a..33b6628 100644 --- a/server_test.go +++ b/server_test.go @@ -306,7 +306,7 @@ func TestConnectionKeepAliveUsesChannelRequestWhenSessionOpen(t *testing.T) { // SendRequest failure (i.e., one probe was wasted on a dead channel // before the prune-on-failure path kicked in). // -// Strategy: use a long ClientAliveInterval (1s) so we can observe the +// Strategy: use a long ClientAliveInterval (3s) so we can observe the // state of openChans in the window between channel close and the first // post-close probe. We capture the openChannelSet from inside the // session handler so the test can inspect it without exporting helpers. @@ -325,7 +325,7 @@ func TestConnectionKeepAlivePrunesClosedChannels(t *testing.T) { } <-ctx.Done() }, - ClientAliveInterval: 1 * time.Second, + ClientAliveInterval: 3 * time.Second, ClientAliveCountMax: 100, // generous so the conn isn't torn down mid-test } @@ -414,12 +414,18 @@ func TestConnectionKeepAlivePrunesClosedChannels(t *testing.T) { <-chReqWg[i] } - // Poll for the close hook to drain. ClientAliveInterval is 1s, so we + // Poll for the close hook to drain. ClientAliveInterval is 3s, so we // have ample time to observe the prune BEFORE the next probe fires. // If the close hook works, len(openChans.chans) drops to 0 quickly // (just goroutine scheduling). If the hook is broken, it stays at 3 // until the next keepalive tick fires and SendRequest fails. - deadline := time.Now().Add(800 * time.Millisecond) + // + // We use a 500ms poll deadline — well above goroutine-scheduling + // jitter but well below the 3s tick interval. This guarantees the + // assertion is unambiguous: it passes ONLY if the close hook ran, + // not because a keepalive tick happened to fire and the + // prune-on-failure path cleared the set. + deadline := time.Now().Add(500 * time.Millisecond) for time.Now().Before(deadline) { openChans.mu.Lock() n := len(openChans.chans) @@ -432,18 +438,29 @@ func TestConnectionKeepAlivePrunesClosedChannels(t *testing.T) { openChans.mu.Lock() regAfter := len(openChans.chans) openChans.mu.Unlock() + // Invariant: no keepalive tick should have fired during the 500ms + // window (interval is 3s). If one did, the test result is + // ambiguous — the prune may have happened via the + // SendRequest-failure path instead of the close hook. + if globalCount.Load() != 0 { + t.Fatalf( + "keepalive tick fired before assertion window (globalCount=%d); test invalid", + globalCount.Load(), + ) + } if regAfter != 0 { t.Fatalf( - "openChannelSet still has %d entries 800ms after close; close hook did not run", + "openChannelSet still has %d entries 500ms after close; close hook did not run", regAfter, ) } - // And verify that the next keepalive probe (which fires ~1s after the - // last reset, i.e. shortly after this point) goes out as a global - // request — confirming behavior end-to-end. + // And verify that the next keepalive probe (which fires ~3s after the + // last reset) goes out as a global request — confirming behavior + // end-to-end. Wait 4s to leave room for at least one tick after the + // 3s interval. before := globalCount.Load() - time.Sleep(1500 * time.Millisecond) + time.Sleep(4 * time.Second) if got := globalCount.Load() - before; got < 1 { t.Fatalf("expected >=1 global-typed keepalive after channels closed, got %d", got) } From c8df8debd52409b2efb781c250ba26610228b4d9 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 25 May 2026 16:10:49 -0500 Subject: [PATCH 07/10] fix(keepalive): drop unsafe defer keepAlive.Close, doc improvements MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Removed defer keepAlive.Close() in connectionKeepAlive. The SessionKeepAlive is referenced by other goroutines (handleRequests, per-channel forwarders, in-flight probes) that may still call NotePeerActivity/Reset after connectionKeepAlive returns. The closed guards prevented an immediate panic but the pattern was fragile. The ticker is unreferenced once HandleConn returns and GC reclaims it. - ConnectionClosingCallback godoc now warns that channel handler goroutines may still be live when the callback fires; points to ConnectionCompleteCallback for all-work-done semantics. - Unexported the context key for openChannelSet (was ContextKeyOpenChannels, now contextKeyOpenChannels) — the value type is unexported so the exported key was misleading. - Trimmed NotePeerActivity godoc to drop the file-specific OpenSSH source citation while preserving the high-level semantic claim. --- context.go | 8 +++++--- keepalive.go | 16 ++++++---------- server.go | 11 ++++++++--- server_test.go | 2 +- ssh.go | 7 +++++++ 5 files changed, 27 insertions(+), 17 deletions(-) diff --git a/context.go b/context.go index 60cc781..07213a7 100644 --- a/context.go +++ b/context.go @@ -61,9 +61,11 @@ var ( // The associated value will be of type *SessionKeepAlive. ContextKeyKeepAlive = &contextKey{"keep-alive"} - // ContextKeyOpenChannels is a context key for use with Contexts in this package. - // The associated value will be of type *openChannelSet. - ContextKeyOpenChannels = &contextKey{"open-channels"} + // contextKeyOpenChannels is an internal context key for the + // per-connection *openChannelSet used by connection keep-alive + // bookkeeping. Unexported because the value type is unexported and + // external consumers should not depend on this detail. + contextKeyOpenChannels = &contextKey{"open-channels"} ) // Context is a package specific context interface. It exposes connection diff --git a/keepalive.go b/keepalive.go index a407484..9a36ac0 100644 --- a/keepalive.go +++ b/keepalive.go @@ -71,16 +71,12 @@ func (ska *SessionKeepAlive) Reset() { } } -// NotePeerActivity records that inbound traffic was observed from the -// peer. Like Reset, it bumps lastReceived (which clears the dead-peer -// deadline used by TimeIsUp) and resets the ticker (suppressing a -// redundant probe). Unlike Reset, it does NOT bump the -// KeepAliveReplyReceived metric — that's reserved for actual probe -// replies. This matches OpenSSH sshd: any successfully-received packet -// zeros keep_alive_timeouts (packet.c ssh_packet_read_poll_seqnr) and -// defers the next probe (serverloop.c). The unanswered-probe counter -// is a function of silence on the wire, not of probe-reply tracking -// specifically. +// NotePeerActivity records inbound traffic from the peer. It bumps +// lastReceived (clearing the dead-peer deadline used by TimeIsUp) and +// resets the ticker so the next probe fires `interval` after the most +// recent activity. Matches OpenSSH sshd, which clears keep_alive_timeouts +// on every successfully-received packet and defers the next probe on +// inbound traffic. func (ska *SessionKeepAlive) NotePeerActivity() { ska.m.Lock() defer ska.m.Unlock() diff --git a/server.go b/server.go index f6f355a..fe22047 100644 --- a/server.go +++ b/server.go @@ -454,7 +454,7 @@ func (srv *Server) HandleConn(newConn net.Conn) { // To prevent race conditions, we need to configure the keep-alive before goroutines kick off applyKeepAlive(ctx, srv.ClientAliveInterval, srv.ClientAliveCountMax) openChans := &openChannelSet{} - ctx.SetValue(ContextKeyOpenChannels, openChans) + ctx.SetValue(contextKeyOpenChannels, openChans) // Connection-level keep-alive: runs for the lifetime of the transport, // independent of whether any session is active. This is what detects a @@ -522,10 +522,15 @@ func (srv *Server) connectionKeepAlive( // Reuse the SessionKeepAlive already stashed on ctx so request-handler // resets (KeepAliveRequestHandler) and metrics keep working. + // + // Do NOT Close() the SessionKeepAlive when this function returns: other + // goroutines (Server.handleRequests, trackingNewChannel.Accept's + // forwarder, and any in-flight probe goroutine spawned below) may still + // be calling NotePeerActivity / Reset after we return. The ticker is + // no longer referenced once HandleConn returns and is GC'd. keepAlive := ctx.KeepAlive() - defer keepAlive.Close() - openChans, _ := ctx.Value(ContextKeyOpenChannels).(*openChannelSet) + openChans, _ := ctx.Value(contextKeyOpenChannels).(*openChannelSet) inFlight := make(chan struct{}, 1) for { diff --git a/server_test.go b/server_test.go index 33b6628..b793843 100644 --- a/server_test.go +++ b/server_test.go @@ -317,7 +317,7 @@ func TestConnectionKeepAlivePrunesClosedChannels(t *testing.T) { srv := &Server{ Handler: func(s Session) { ctx := s.Context() - if oc, ok := ctx.Value(ContextKeyOpenChannels).(*openChannelSet); ok { + if oc, ok := ctx.Value(contextKeyOpenChannels).(*openChannelSet); ok { select { case openChansCh <- oc: default: diff --git a/ssh.go b/ssh.go index ed1e604..3b4d98f 100644 --- a/ssh.go +++ b/ssh.go @@ -101,6 +101,13 @@ type ConnectionCompleteCallback func(conn *gossh.ServerConn, err error) // the wait. The Context is the same one threaded through auth and channel // handlers, so per-connection state stashed via ctx.SetValue is reachable // without an external sync.Map keyed by *gossh.ServerConn. +// +// Note: channel handler goroutines spawned by HandleConn may still be +// running and mutating per-connection state when this callback fires. +// The callback is intentionally synchronous and pre-defer; it does NOT +// imply that all per-connection work has finished. For "all work done" +// semantics, use ConnectionCompleteCallback (which runs after +// sshConn.Wait). type ConnectionClosingCallback func(ctx Context, conn *gossh.ServerConn) // Window represents the size of a PTY window. From e16d6530fd1d26eeb0b504b237cf4ec2b64a7727 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 25 May 2026 16:21:08 -0500 Subject: [PATCH 08/10] chore: remove dead nil-guard in connectionKeepAlive openChans is unconditionally created and stashed on ctx in HandleConn before connectionKeepAlive can run, so the comma-ok and nil-check were unreachable. Direct type assertion + dropping the guard simplifies the probe-send sequence. --- server.go | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/server.go b/server.go index fe22047..5641eb9 100644 --- a/server.go +++ b/server.go @@ -530,7 +530,7 @@ func (srv *Server) connectionKeepAlive( // no longer referenced once HandleConn returns and is GC'd. keepAlive := ctx.KeepAlive() - openChans, _ := ctx.Value(contextKeyOpenChannels).(*openChannelSet) + openChans := ctx.Value(contextKeyOpenChannels).(*openChannelSet) inFlight := make(chan struct{}, 1) for { @@ -567,10 +567,7 @@ func (srv *Server) connectionKeepAlive( // hangs forever it will be unblocked when the TimeIsUp // branch closes sshConn. var err error - var ch gossh.Channel - if openChans != nil { - ch = openChans.any() - } + ch := openChans.any() if ch != nil { _, err = ch.SendRequest(keepAliveRequestType, true, nil) if err != nil { From 1bd4e78d89b8f553e5b05e120c1b09e7db0edc91 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 25 May 2026 16:22:29 -0500 Subject: [PATCH 09/10] test: add stalled-with-session keepalive test, unexport notePeerActivity - New TestConnectionKeepAliveClosesStalledConnWithOpenSession exercises the channel-request path of dead-peer detection. Opens a session and has the client drain all request streams without replying; the server should still detect the stall via TimeIsUp and tear the transport down. Complements the existing no-session variant which exercises the global-request path. - Unexported NotePeerActivity -> notePeerActivity. It is an internal hook driven by this package's request loops; external callers have no reason to invoke it. Doing this before v1.2.0 is cut avoids a v2 break later. --- keepalive.go | 6 ++-- server.go | 6 ++-- server_test.go | 75 ++++++++++++++++++++++++++++++++++++++++++++++++++ session.go | 2 +- 4 files changed, 82 insertions(+), 7 deletions(-) diff --git a/keepalive.go b/keepalive.go index 9a36ac0..e675d8d 100644 --- a/keepalive.go +++ b/keepalive.go @@ -71,13 +71,13 @@ func (ska *SessionKeepAlive) Reset() { } } -// NotePeerActivity records inbound traffic from the peer. It bumps +// notePeerActivity records inbound traffic from the peer. It bumps // lastReceived (clearing the dead-peer deadline used by TimeIsUp) and // resets the ticker so the next probe fires `interval` after the most // recent activity. Matches OpenSSH sshd, which clears keep_alive_timeouts // on every successfully-received packet and defers the next probe on -// inbound traffic. -func (ska *SessionKeepAlive) NotePeerActivity() { +// inbound traffic. Internal — driven by the package's request loops. +func (ska *SessionKeepAlive) notePeerActivity() { ska.m.Lock() defer ska.m.Unlock() if ska.ticker != nil && !ska.closed { diff --git a/server.go b/server.go index 5641eb9..833f18b 100644 --- a/server.go +++ b/server.go @@ -485,7 +485,7 @@ func (srv *Server) HandleConn(newConn net.Conn) { onClose: openChans.remove, notePeerActivity: func() { if ka := ctx.KeepAlive(); ka != nil { - ka.NotePeerActivity() + ka.notePeerActivity() } }, ctx: ctx, @@ -526,7 +526,7 @@ func (srv *Server) connectionKeepAlive( // Do NOT Close() the SessionKeepAlive when this function returns: other // goroutines (Server.handleRequests, trackingNewChannel.Accept's // forwarder, and any in-flight probe goroutine spawned below) may still - // be calling NotePeerActivity / Reset after we return. The ticker is + // be calling notePeerActivity / Reset after we return. The ticker is // no longer referenced once HandleConn returns and is GC'd. keepAlive := ctx.KeepAlive() @@ -591,7 +591,7 @@ func (srv *Server) connectionKeepAlive( func (srv *Server) handleRequests(ctx Context, in <-chan *gossh.Request) { for req := range in { if ka := ctx.KeepAlive(); ka != nil { - ka.NotePeerActivity() + ka.notePeerActivity() } handler := srv.RequestHandlers[req.Type] if handler == nil { diff --git a/server_test.go b/server_test.go index b793843..298acc8 100644 --- a/server_test.go +++ b/server_test.go @@ -213,6 +213,81 @@ func TestConnectionKeepAliveClosesStalledConn(t *testing.T) { } } +// TestConnectionKeepAliveClosesStalledConnWithOpenSession is the counterpart +// to TestConnectionKeepAliveClosesStalledConn: when a session channel is open +// (so probes go as channel requests per client_alive_check), and the peer +// stops replying to those probes, the connection-level keep-alive must still +// detect the stall and tear the transport down. This exercises the +// channel-request path of the deadline check; the no-session variant exercises +// the global-request path. +func TestConnectionKeepAliveClosesStalledConnWithOpenSession(t *testing.T) { + t.Parallel() + + closingFired := make(chan struct{}) + + srv := &Server{ + Handler: func(s Session) { <-s.Context().Done() }, + ClientAliveInterval: 100 * time.Millisecond, + ClientAliveCountMax: 3, + ConnectionClosingCallback: func(_ Context, _ *gossh.ServerConn) { + close(closingFired) + }, + } + + l := newLocalTCPListener() + defer func() { _ = l.Close() }() + go func() { _ = srv.serveOnce(l) }() + + cfg := &gossh.ClientConfig{ + User: "testuser", + Auth: []gossh.AuthMethod{gossh.Password("testpass")}, + HostKeyCallback: gossh.InsecureIgnoreHostKey(), //nolint:gosec // test code + } + netConn, err := net.Dial("tcp", l.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + sshConn, chans, reqs, err := gossh.NewClientConn(netConn, l.Addr().String(), cfg) + if err != nil { + t.Fatalf("NewClientConn: %v", err) + } + defer func() { _ = sshConn.Close() }() + + ch, chReqs, err := sshConn.OpenChannel("session", nil) + if err != nil { + t.Fatalf("OpenChannel: %v", err) + } + defer func() { _ = ch.Close() }() + + // Drain all three request streams (global, per-channel, and the + // NewChannel stream) WITHOUT replying. The server's channel-typed + // keepalive probes go on chReqs and will never get a reply, so + // notePeerActivity is never called on the session-request side and + // TimeIsUp eventually fires. + go func() { + for range chans { //nolint:revive // intentional drain + } + }() + go func() { + for req := range reqs { + _ = req + } + }() + go func() { + for req := range chReqs { + _ = req + } + }() + + // 100ms * 3 = 300ms expected; allow generous slack for CI. + select { + case <-closingFired: + case <-time.After(5 * time.Second): + t.Fatal("ConnectionClosingCallback did not fire; " + + "stalled-with-session client was not torn down by connection-level keep-alive") + } +} + // TestConnectionKeepAliveUsesChannelRequestWhenSessionOpen verifies that the // server's connection-level keepalive mirrors OpenSSH's client_alive_check(): // while at least one channel is open it sends keepalive@openssh.com as a diff --git a/session.go b/session.go index 138ba44..2b9d56e 100644 --- a/session.go +++ b/session.go @@ -294,7 +294,7 @@ func (sess *session) handleRequests(_ Context, reqs <-chan *gossh.Request) { }() for req := range reqs { if ka := sess.ctx.KeepAlive(); ka != nil { - ka.NotePeerActivity() + ka.notePeerActivity() } switch req.Type { case "shell", "exec": From 8e95dba915410cb7f71ea9efa904edbf9250a541 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 25 May 2026 16:39:01 -0500 Subject: [PATCH 10/10] fix(keepalive): deterministically Stop ticker on HandleConn teardown Per independent review: rely on GC was option A; explicit Stop in HandleConn's defer chain is option B. The ticker stays reachable after HandleConn returns because ctx is held by spawned handler goroutines and the per-channel request forwarder, so it kept firing until those exited. LIFO ensures close(keepAliveDone) runs first (signaling connectionKeepAlive to return), then ctx.KeepAlive().Close() Stops the ticker. All SessionKeepAlive methods are mutex-protected and check closed before touching the ticker, so a straggler caller is a safe no-op. --- server.go | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/server.go b/server.go index 833f18b..a95acc3 100644 --- a/server.go +++ b/server.go @@ -465,6 +465,15 @@ func (srv *Server) HandleConn(newConn net.Conn) { // callbacks can fire. keepAliveDone := make(chan struct{}) go srv.connectionKeepAlive(ctx, sshConn, keepAliveDone) + // LIFO: close(keepAliveDone) runs first to signal connectionKeepAlive + // to return, then KeepAlive().Close() stops the ticker deterministically. + // Without the explicit Close, the ticker keeps firing on a dropped + // channel until ctx (held by spawned handler goroutines and the + // per-channel forwarder) becomes unreachable and GC reclaims it. + // All SessionKeepAlive methods are mutex-protected, so a straggler + // goroutine calling notePeerActivity/Reset after Close just sees + // closed=true and no-ops on the ticker. + defer ctx.KeepAlive().Close() defer close(keepAliveDone) // go gossh.DiscardRequests(reqs) @@ -521,13 +530,9 @@ func (srv *Server) connectionKeepAlive( } // Reuse the SessionKeepAlive already stashed on ctx so request-handler - // resets (KeepAliveRequestHandler) and metrics keep working. - // - // Do NOT Close() the SessionKeepAlive when this function returns: other - // goroutines (Server.handleRequests, trackingNewChannel.Accept's - // forwarder, and any in-flight probe goroutine spawned below) may still - // be calling notePeerActivity / Reset after we return. The ticker is - // no longer referenced once HandleConn returns and is GC'd. + // resets (KeepAliveRequestHandler) and metrics keep working. The ticker + // is Stop()ped from HandleConn's defer chain after this goroutine has + // been signaled to return. keepAlive := ctx.KeepAlive() openChans := ctx.Value(contextKeyOpenChannels).(*openChannelSet)