From 3af65125147e4fd7236e353e495406d4bf1316fc Mon Sep 17 00:00:00 2001 From: Aditya Raj Singh Date: Sun, 30 Aug 2026 23:04:32 +0530 Subject: [PATCH] fix(hub): don't read the SSH client after it is closed (#2277) createSessionWithTimeout checked sys.client for nil and then dereferenced it again inside the goroutine that calls NewSession. update() runs the SMART fetch in its own goroutine, so closeSSHConnection can clear the field between those two reads and the goroutine dereferences a nil client, panicking the whole hub process. Make client an atomic.Pointer, load it once before starting the goroutine, and clear it with Swap so a concurrent close cannot be observed mid-session-creation. NewSession on an already-closed client returns an error, which the existing retry path already handles. Closes #2157 --- internal/hub/systems/ssh_timeout_test.go | 37 ++++++++++++++++ internal/hub/systems/system.go | 56 ++++++++++++------------ 2 files changed, 65 insertions(+), 28 deletions(-) diff --git a/internal/hub/systems/ssh_timeout_test.go b/internal/hub/systems/ssh_timeout_test.go index 2e7e1373..d0b3c671 100644 --- a/internal/hub/systems/ssh_timeout_test.go +++ b/internal/hub/systems/ssh_timeout_test.go @@ -4,11 +4,13 @@ package systems import ( "errors" + "sync" "testing" "testing/synctest" "time" "github.com/stretchr/testify/assert" + "golang.org/x/crypto/ssh" ) // TestRunWithTimeout covers the guard added for issue #2041: the per-system SSH @@ -54,3 +56,38 @@ func TestRunWithTimeout(t *testing.T) { }) }) } + +// closedConn stands in for a connection whose peer has gone away: opening a +// channel fails rather than succeeding, which is what NewSession does on a +// client that closeSSHConnection has already closed. +type closedConn struct{ ssh.Conn } + +func (closedConn) OpenChannel(string, []byte) (ssh.Channel, <-chan *ssh.Request, error) { + return nil, nil, errors.New("use of closed network connection") +} + +func (closedConn) Close() error { return nil } + +// TestCreateSessionDuringClose covers issue #2157: the background SMART fetch +// creates a session while the updater can be tearing the same connection down, +// so session creation must not read the client field after it is cleared. +func TestCreateSessionDuringClose(t *testing.T) { + for range 500 { + sys := &System{ctx: t.Context()} + sys.client.Store(&ssh.Client{Conn: closedConn{}}) + + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + session, err := sys.createSessionWithTimeout(time.Second) + assert.Nil(t, session) + assert.Error(t, err, "a closed connection must surface an error, not a session") + }() + go func() { + defer wg.Done() + sys.closeSSHConnection() + }() + wg.Wait() + } +} diff --git a/internal/hub/systems/system.go b/internal/hub/systems/system.go index c04c2b78..6c0879a1 100644 --- a/internal/hub/systems/system.go +++ b/internal/hub/systems/system.go @@ -33,22 +33,22 @@ import ( ) type System struct { - Id string `db:"id"` - Host string `db:"host"` - Port string `db:"port"` - Status string `db:"status"` - manager *SystemManager // Manager that this system belongs to - client *ssh.Client // SSH client for fetching data - sshTransport *transport.SSHTransport // SSH transport for requests - data *system.CombinedData // system data from agent - ctx context.Context // Context for stopping the updater - cancel context.CancelFunc // Stops and removes system from updater - WsConn *ws.WsConn // Handler for agent WebSocket connection - agentVersion semver.Version // Agent version - updateTicker *time.Ticker // Ticker for updating the system - detailsFetched atomic.Bool // True if static system details have been fetched and saved - smartFetching atomic.Bool // True if SMART devices are currently being fetched - smartInterval time.Duration // Interval for periodic SMART data updates + Id string `db:"id"` + Host string `db:"host"` + Port string `db:"port"` + Status string `db:"status"` + manager *SystemManager // Manager that this system belongs to + client atomic.Pointer[ssh.Client] // SSH client for fetching data + sshTransport *transport.SSHTransport // SSH transport for requests + data *system.CombinedData // system data from agent + ctx context.Context // Context for stopping the updater + cancel context.CancelFunc // Stops and removes system from updater + WsConn *ws.WsConn // Handler for agent WebSocket connection + agentVersion semver.Version // Agent version + updateTicker *time.Ticker // Ticker for updating the system + detailsFetched atomic.Bool // True if static system details have been fetched and saved + smartFetching atomic.Bool // True if SMART devices are currently being fetched + smartInterval time.Duration // Interval for periodic SMART data updates } func (sm *SystemManager) NewSystem(systemId string) *System { @@ -434,7 +434,7 @@ func (sys *System) request(ctx context.Context, action common.WebSocketAction, r err := sys.sshTransport.RequestWithRetry(ctx, action, req, dest, 1) // Keep legacy SSH client/version fields in sync for other code paths. if sys.sshTransport != nil { - sys.client = sys.sshTransport.GetClient() + sys.client.Store(sys.sshTransport.GetClient()) sys.agentVersion = sys.sshTransport.GetAgentVersion() } return err @@ -476,8 +476,8 @@ func (sys *System) ensureSSHTransport() error { }) } // Sync client state with transport - if sys.client != nil { - sys.sshTransport.SetClient(sys.client) + if client := sys.client.Load(); client != nil { + sys.sshTransport.SetClient(client) sys.sshTransport.SetAgentVersion(sys.agentVersion) } return nil @@ -625,7 +625,7 @@ func (sys *System) fetchDataViaSSH(options common.DataRequestOptions) (*system.C // The operation can request a retry by returning true as the first return value. func (sys *System) runSSHOperation(timeout time.Duration, retries int, operation func(*ssh.Session) (bool, error)) error { for attempt := 0; attempt <= retries; attempt++ { - if sys.client == nil || sys.Status == down { + if sys.client.Load() == nil || sys.Status == down { if err := sys.createSSHClient(); err != nil { return err } @@ -721,12 +721,12 @@ func (s *System) createSSHClient() error { } else { host = net.JoinHostPort(host, s.Port) } - var err error - s.client, err = dialSSHWithKeepAlive(network, host, s.manager.sshConfig) + client, err := dialSSHWithKeepAlive(network, host, s.manager.sshConfig) + s.client.Store(client) if err != nil { return err } - s.agentVersion, _ = extractAgentVersion(string(s.client.Conn.ServerVersion())) + s.agentVersion, _ = extractAgentVersion(string(client.Conn.ServerVersion())) s.manager.resetFailedSmartFetchState(s.Id) return nil } @@ -762,7 +762,8 @@ func dialSSHWithKeepAlive(network, addr string, config *ssh.ClientConfig) (*ssh. // createSessionWithTimeout creates a new SSH session with a timeout to avoid hanging // in case of network issues func (sys *System) createSessionWithTimeout(timeout time.Duration) (*ssh.Session, error) { - if sys.client == nil { + client := sys.client.Load() + if client == nil { return nil, fmt.Errorf("client not initialized") } @@ -773,7 +774,7 @@ func (sys *System) createSessionWithTimeout(timeout time.Duration) (*ssh.Session errChan := make(chan error, 1) go func() { - if session, err := sys.client.NewSession(); err != nil { + if session, err := client.NewSession(); err != nil { errChan <- err } else { sessionChan <- session @@ -795,9 +796,8 @@ func (sys *System) closeSSHConnection() { if sys.sshTransport != nil { sys.sshTransport.Close() } - if sys.client != nil { - sys.client.Close() - sys.client = nil + if client := sys.client.Swap(nil); client != nil { + client.Close() } }