mirrors/beszel
1
0
Fork 0
mirror of https://github.com/henrygd/beszel.git synced 2026-09-19 14:34:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(hub): don't read the SSH client after it is closed (#2277)

Browse Source 
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
This commit is contained in:
Aditya Raj Singh
2026-08-30 13:34:32 -04:00
committed by GitHub
parent 87620f3251
commit 3af6512514
2 changed files with 65 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/hub/systems/ssh_timeout_test.go
+37
View File
@@ -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()
}
}
internal/hub/systems/system.go
+28 -28
View File
@@ -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()
}
}