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() } }