From bfa82df84ba55f2b86c3dd9ded6760282ca4ee9f Mon Sep 17 00:00:00 2001 From: Josh Keldam Date: Thu, 5 Nov 2026 17:06:02 +0900 Subject: [PATCH 1/2] fix(initiator): bound writes on stop and close the connection A socket write has no deadline, the session loop waits for the writer to take each message, and Initiator.Stop waits for the session loop. When the counterparty stops reading, Stop either hangs (the writer was already blocked, so the session loop cannot take the stop) or returns while the connection is still open (the writer is blocked on the Logout itself, after the session timed the Logout out). Only the writer closes the connection, so nothing else ever does. Once the initiator stops, every write, including one already in progress, now has LogoutTimeout to finish, set again for each write. A write that misses it closes the connection, which frees the writer and the session loop. After the session has stopped, the writer gets LogoutTimeout to finish its last message, so a graceful Logout is not cut short, and then the transport under any TLS is closed, so Stop never returns with the connection open. Tests cover a peer that stops reading with the socket buffers full, over TCP and TLS (Stop hangs without this change), a peer that answers the Logout (still graceful), and a write blocked on the Logout itself. --- connection.go | 55 +++++- initiator.go | 24 ++- initiator_test.go | 472 ++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 549 insertions(+), 2 deletions(-) create mode 100644 initiator_test.go diff --git a/connection.go b/connection.go index 99a4c465e..9f0886d68 100644 --- a/connection.go +++ b/connection.go @@ -15,7 +15,60 @@ package quickfix -import "io" +import ( + "crypto/tls" + "errors" + "io" + "net" + "sync/atomic" + "time" +) + +// stopWriter is the initiator's connection as its writer sees it. Once the +// initiator stops, every write, including one already in progress, must +// finish within timeout. A write that does not leaves the stream with a +// partial message on it and the counterparty not reading, so it closes the +// connection: the writer then fails at once, and the session can stop. +type stopWriter struct { + conn net.Conn + timeout time.Duration + stopping atomic.Bool +} + +// stop bounds the write in progress and every later one by timeout. +func (w *stopWriter) stop() { + w.stopping.Store(true) + _ = w.conn.SetWriteDeadline(time.Now().Add(w.timeout)) +} + +func (w *stopWriter) Write(b []byte) (int, error) { + if w.stopping.Load() { + _ = w.conn.SetWriteDeadline(time.Now().Add(w.timeout)) + } + n, err := w.conn.Write(b) + if err != nil && w.stopping.Load() { + _ = transport(w.conn).Close() + } + return n, err +} + +// transport is the connection under TLS, if any. Closing it ends a blocked +// write or close notification at once; closing a tls.Conn already being +// closed only reports net.ErrClosed while that close notification waits. +func transport(conn net.Conn) net.Conn { + if tlsConn, ok := conn.(*tls.Conn); ok { + return tlsConn.NetConn() + } + return conn +} + +// closeTransport closes the connection's transport. It is closed once this +// returns, whoever closed it first. +func closeTransport(conn net.Conn, log Log) { + if err := transport(conn).Close(); err != nil && !errors.Is(err, net.ErrClosed) { + log.OnEvent(err.Error()) + } +} func writeLoop(connection io.Writer, messageOut chan []byte, log Log) { for { diff --git a/initiator.go b/initiator.go index 9af90cb2b..fd54d7240 100644 --- a/initiator.go +++ b/initiator.go @@ -179,6 +179,7 @@ func (i *Initiator) handleConnection(session *session, tlsConfig *tls.Config, di var disconnected chan interface{} var msgIn chan fixIn var msgOut chan []byte + var writer *stopWriter address := session.SocketConnectAddress[connectionAttempt%len(session.SocketConnectAddress)] session.log.OnEventf("Connecting to: %v", address) @@ -214,8 +215,9 @@ func (i *Initiator) handleConnection(session *session, tlsConfig *tls.Config, di go readLoop(newParser(bufio.NewReader(netConn)), msgIn, session.log) disconnected = make(chan interface{}) + writer = &stopWriter{conn: netConn, timeout: session.LogoutTimeout} go func() { - writeLoop(netConn, msgOut, session.log) + writeLoop(writer, msgOut, session.log) if err := netConn.Close(); err != nil { session.log.OnEvent(err.Error()) } @@ -229,6 +231,26 @@ func (i *Initiator) handleConnection(session *session, tlsConfig *tls.Config, di select { case <-disconnected: case <-i.stopChan: + // A write has no deadline of its own, and the session loop + // waits for the writer to take each message: a counterparty + // that stops reading would hold a write, the session loop + // behind it, and so this stop, with the connection open. From + // here every write gets LogoutTimeout. A write still not taken + // then closes the connection, and the session can stop. + writer.stop() + session.stop() + wg.Wait() + // The writer may still be on its last message (a Logout, or + // the answer to the counterparty's) or closing the connection. + // Give it LogoutTimeout, then close the transport under it, so + // Stop never returns while the connection is open. + timer := time.NewTimer(session.LogoutTimeout) + select { + case <-disconnected: + case <-timer.C: + } + timer.Stop() + closeTransport(netConn, session.log) return } diff --git a/initiator_test.go b/initiator_test.go new file mode 100644 index 000000000..cb22dac8a --- /dev/null +++ b/initiator_test.go @@ -0,0 +1,472 @@ +// Copyright (c) quickfixengine.org All rights reserved. +// +// This file may be distributed under the terms of the quickfixengine.org +// license as defined by quickfixengine.org and appearing in the file +// LICENSE included in the packaging of this file. +// +// This file is provided AS IS with NO WARRANTY OF ANY KIND, INCLUDING +// THE WARRANTY OF DESIGN, MERCHANTABILITY AND FITNESS FOR A +// PARTICULAR PURPOSE. +// +// See http://www.quickfixengine.org/LICENSE for licensing information. +// +// Contact ask@quickfixengine.org if any conditions of this licensing +// are not clear to you. + +package quickfix + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "errors" + "fmt" + "io" + "math/big" + "net" + "os" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// stallingConn stands in for a connection to a counterparty that stops +// reading: once stalled, every write blocks until the connection is closed. +type stallingConn struct { + net.Conn + stalled atomic.Bool + closed chan struct{} + closeOnce sync.Once +} + +func (c *stallingConn) Write(b []byte) (int, error) { + if c.stalled.Load() { + <-c.closed + return 0, net.ErrClosed + } + return c.Conn.Write(b) +} + +func (c *stallingConn) Close() error { + c.closeOnce.Do(func() { close(c.closed) }) + return c.Conn.Close() +} + +func (c *stallingConn) isClosed() bool { + select { + case <-c.closed: + return true + default: + return false + } +} + +type stallingDialer struct { + conns chan *stallingConn +} + +func (d stallingDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) { + var dialer net.Dialer + conn, err := dialer.DialContext(ctx, network, address) + if err != nil { + return nil, err + } + stalling := &stallingConn{Conn: conn, closed: make(chan struct{})} + d.conns <- stalling + return stalling, nil +} + +type logonRecorder struct { + loggedOn chan SessionID + loggedOut chan SessionID +} + +func (a logonRecorder) OnCreate(SessionID) {} +func (a logonRecorder) OnLogon(sessionID SessionID) { + select { + case a.loggedOn <- sessionID: + default: + } +} +func (a logonRecorder) OnLogout(sessionID SessionID) { + select { + case a.loggedOut <- sessionID: + default: + } +} +func (a logonRecorder) ToAdmin(*Message, SessionID) {} +func (a logonRecorder) ToApp(*Message, SessionID) error { return nil } +func (a logonRecorder) FromAdmin(*Message, SessionID) MessageRejectError { return nil } +func (a logonRecorder) FromApp(*Message, SessionID) MessageRejectError { return nil } + +// A counterparty that stops reading leaves the writer blocked on the Logout +// with no deadline, while the session loop times the Logout out and exits. +// Stop must not return with that connection still open: the caller takes a +// returned Stop to mean the session is gone, and may log the same session on +// again. +func TestInitiatorStopClosesAConnectionBlockedOnLogout(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + port := listener.Addr().(*net.TCPAddr).Port + require.NoError(t, listener.Close()) + + acceptorSettings, err := ParseSettings(strings.NewReader(fmt.Sprintf(`[DEFAULT] +SocketAcceptPort=%d +[SESSION] +BeginString=FIX.4.4 +SenderCompID=STALL_ACCEPTOR +TargetCompID=STALL_INITIATOR +HeartBtInt=30 +`, port))) + require.NoError(t, err) + acceptor, err := NewAcceptor(logonRecorder{loggedOn: make(chan SessionID, 1)}, NewMemoryStoreFactory(), acceptorSettings, NewNullLogFactory()) + require.NoError(t, err) + require.NoError(t, acceptor.Start()) + t.Cleanup(acceptor.Stop) + + initiatorSettings, err := ParseSettings(strings.NewReader(fmt.Sprintf(`[DEFAULT] +ReconnectInterval=1 +LogoutTimeout=1 +[SESSION] +BeginString=FIX.4.4 +SenderCompID=STALL_INITIATOR +TargetCompID=STALL_ACCEPTOR +SocketConnectHost=127.0.0.1 +SocketConnectPort=%d +HeartBtInt=30 +`, port))) + require.NoError(t, err) + app := logonRecorder{loggedOn: make(chan SessionID, 1)} + initiator, err := NewInitiator(app, NewMemoryStoreFactory(), initiatorSettings, NewNullLogFactory()) + require.NoError(t, err) + + // Start, with a dialer whose connection the test can stall. + dialer := stallingDialer{conns: make(chan *stallingConn, 1)} + initiator.stopChan = make(chan interface{}) + for sessionID := range initiator.sessionSettings { + initiator.wg.Add(1) + go func(sessionID SessionID) { + defer initiator.wg.Done() + initiator.handleConnection(initiator.sessions[sessionID], nil, dialer) + }(sessionID) + } + + conn := <-dialer.conns + select { + case <-app.loggedOn: + case <-time.After(10 * time.Second): + t.Fatal("the initiator did not log on") + } + conn.stalled.Store(true) + + stopped := make(chan struct{}) + go func() { + initiator.Stop() + close(stopped) + }() + select { + case <-stopped: + case <-time.After(10 * time.Second): + conn.Close() // let the stuck Stop end with the test + t.Fatal("Stop did not return") + } + require.True(t, conn.isClosed(), "Stop returned with the connection open") +} + +// stallingProxy forwards between an initiator and an acceptor, terminating +// TLS on the initiator's side when it has a certificate. Stalled, it stops +// reading what the initiator sends, so the initiator's writes block once the +// socket buffers fill. +type stallingProxy struct { + listener net.Listener + initiator chan net.Conn + stalled atomic.Bool +} + +func startStallingProxy(t *testing.T, acceptorPort int, cert *tls.Certificate) *stallingProxy { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + p := &stallingProxy{listener: listener, initiator: make(chan net.Conn, 1)} + t.Cleanup(func() { _ = listener.Close() }) + go func() { + raw, err := listener.Accept() + if err != nil { + return + } + initiator := raw + if cert != nil { + initiator = tls.Server(raw, &tls.Config{Certificates: []tls.Certificate{*cert}}) + } + acceptor, err := net.Dial("tcp", fmt.Sprintf("127.0.0.1:%d", acceptorPort)) + if err != nil { + _ = initiator.Close() + return + } + t.Cleanup(func() { + _ = initiator.Close() + _ = acceptor.Close() + }) + p.initiator <- initiator + go func() { _, _ = io.Copy(initiator, acceptor) }() + buf := make([]byte, 32*1024) + for !p.stalled.Load() { + _ = initiator.SetReadDeadline(time.Now().Add(10 * time.Millisecond)) + n, err := initiator.Read(buf) + if n > 0 { + if _, werr := acceptor.Write(buf[:n]); werr != nil { + return + } + } + if err != nil && !errors.Is(err, os.ErrDeadlineExceeded) { + return + } + } + }() + return p +} + +func selfSignedCertificate(t *testing.T) (*tls.Certificate, *x509.CertPool) { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + template := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "127.0.0.1"}, + IPAddresses: []net.IP{net.ParseIP("127.0.0.1")}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + } + der, err := x509.CreateCertificate(rand.Reader, template, template, &key.PublicKey, key) + require.NoError(t, err) + parsed, err := x509.ParseCertificate(der) + require.NoError(t, err) + roots := x509.NewCertPool() + roots.AddCert(parsed) + return &tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key}, roots +} + +type plainDialer struct{} + +func (plainDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) { + var dialer net.Dialer + return dialer.DialContext(ctx, network, address) +} + +// A counterparty that stops reading while the session still has messages +// queued: the writer is already blocked when Stop is called, so the session +// loop is blocked behind it and cannot take the stop. Stop must still return +// within its timeouts, with the connection closed, over plain TCP and TLS. +func TestInitiatorStopEndsAWriteAlreadyBlocked(t *testing.T) { + for _, useTLS := range []bool{false, true} { + name := "tcp" + if useTLS { + name = "tls" + } + t.Run(name, func(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + acceptorPort := listener.Addr().(*net.TCPAddr).Port + require.NoError(t, listener.Close()) + acceptorID, initiatorID := "FULL_ACCEPTOR_"+name, "FULL_INITIATOR_"+name + + acceptorSettings, err := ParseSettings(strings.NewReader(fmt.Sprintf(`[DEFAULT] +SocketAcceptPort=%d +[SESSION] +BeginString=FIX.4.4 +SenderCompID=%s +TargetCompID=%s +HeartBtInt=30 +`, acceptorPort, acceptorID, initiatorID))) + require.NoError(t, err) + acceptor, err := NewAcceptor(logonRecorder{loggedOn: make(chan SessionID, 1)}, NewMemoryStoreFactory(), acceptorSettings, NewNullLogFactory()) + require.NoError(t, err) + require.NoError(t, acceptor.Start()) + t.Cleanup(acceptor.Stop) + + var cert *tls.Certificate + var tlsConfig *tls.Config + if useTLS { + var roots *x509.CertPool + cert, roots = selfSignedCertificate(t) + tlsConfig = &tls.Config{RootCAs: roots, MinVersion: tls.VersionTLS12} + } + proxy := startStallingProxy(t, acceptorPort, cert) + + initiatorSettings, err := ParseSettings(strings.NewReader(fmt.Sprintf(`[DEFAULT] +ReconnectInterval=1 +LogoutTimeout=1 +[SESSION] +BeginString=FIX.4.4 +SenderCompID=%s +TargetCompID=%s +SocketConnectHost=127.0.0.1 +SocketConnectPort=%d +HeartBtInt=30 +`, initiatorID, acceptorID, proxy.listener.Addr().(*net.TCPAddr).Port))) + require.NoError(t, err) + app := logonRecorder{loggedOn: make(chan SessionID, 1)} + initiator, err := NewInitiator(app, NewMemoryStoreFactory(), initiatorSettings, NewNullLogFactory()) + require.NoError(t, err) + initiator.stopChan = make(chan interface{}) + var sessionID SessionID + for id := range initiator.sessionSettings { + sessionID = id + initiator.wg.Add(1) + go func() { + defer initiator.wg.Done() + initiator.handleConnection(initiator.sessions[id], tlsConfig, plainDialer{}) + }() + } + + var proxied net.Conn + select { + case proxied = <-proxy.initiator: + case <-time.After(10 * time.Second): + t.Fatal("the initiator did not connect") + } + select { + case <-app.loggedOn: + case <-time.After(10 * time.Second): + t.Fatal("the initiator did not log on") + } + + proxy.stalled.Store(true) + filler := strings.Repeat("x", 64*1024) + for range 512 { // 32 MiB, past any loopback socket buffers + msg := NewMessage() + msg.Header.SetString(tagMsgType, "1") // TestRequest + msg.Body.SetString(Tag(112), filler) + require.NoError(t, SendToTarget(msg, sessionID)) + } + time.Sleep(500 * time.Millisecond) // the writer blocks on a full socket + + stopped := make(chan struct{}) + go func() { + initiator.Stop() + close(stopped) + }() + select { + case <-stopped: + case <-time.After(6 * time.Second): // LogoutTimeout is 1 s + _ = proxied.Close() // let the stuck Stop end with the test + t.Fatal("Stop did not return") + } + + // The proxy drains what was sent and then reaches the end of + // the connection: the initiator closed it. + done := make(chan error, 1) + go func() { + _ = proxied.SetReadDeadline(time.Now().Add(10 * time.Second)) + _, err := io.Copy(io.Discard, proxied) + done <- err + }() + // A close without TLS close notification reads as an + // unexpected EOF; only a deadline means the connection is open. + if err := <-done; errors.Is(err, os.ErrDeadlineExceeded) { + t.Fatalf("the initiator's connection was not closed: %v", err) + } + }) + } +} + +// With a counterparty that reads and answers, Stop still logs out +// gracefully: the counterparty receives the Logout and answers it before the +// connection closes, over plain TCP and TLS. +func TestInitiatorStopLogsOutGracefully(t *testing.T) { + for _, useTLS := range []bool{false, true} { + name := "tcp" + if useTLS { + name = "tls" + } + t.Run(name, func(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + acceptorPort := listener.Addr().(*net.TCPAddr).Port + require.NoError(t, listener.Close()) + acceptorID, initiatorID := "GRACE_ACCEPTOR_"+name, "GRACE_INITIATOR_"+name + + acceptorSettings, err := ParseSettings(strings.NewReader(fmt.Sprintf(`[DEFAULT] +SocketAcceptPort=%d +[SESSION] +BeginString=FIX.4.4 +SenderCompID=%s +TargetCompID=%s +HeartBtInt=30 +`, acceptorPort, acceptorID, initiatorID))) + require.NoError(t, err) + counterparty := logonRecorder{loggedOn: make(chan SessionID, 1), loggedOut: make(chan SessionID, 1)} + acceptor, err := NewAcceptor(counterparty, NewMemoryStoreFactory(), acceptorSettings, NewNullLogFactory()) + require.NoError(t, err) + require.NoError(t, acceptor.Start()) + t.Cleanup(acceptor.Stop) + + var cert *tls.Certificate + var tlsConfig *tls.Config + if useTLS { + var roots *x509.CertPool + cert, roots = selfSignedCertificate(t) + tlsConfig = &tls.Config{RootCAs: roots, MinVersion: tls.VersionTLS12} + } + proxy := startStallingProxy(t, acceptorPort, cert) + + initiatorSettings, err := ParseSettings(strings.NewReader(fmt.Sprintf(`[DEFAULT] +ReconnectInterval=1 +LogoutTimeout=5 +[SESSION] +BeginString=FIX.4.4 +SenderCompID=%s +TargetCompID=%s +SocketConnectHost=127.0.0.1 +SocketConnectPort=%d +HeartBtInt=30 +`, initiatorID, acceptorID, proxy.listener.Addr().(*net.TCPAddr).Port))) + require.NoError(t, err) + app := logonRecorder{loggedOn: make(chan SessionID, 1), loggedOut: make(chan SessionID, 1)} + initiator, err := NewInitiator(app, NewMemoryStoreFactory(), initiatorSettings, NewNullLogFactory()) + require.NoError(t, err) + initiator.stopChan = make(chan interface{}) + for id := range initiator.sessionSettings { + initiator.wg.Add(1) + go func() { + defer initiator.wg.Done() + initiator.handleConnection(initiator.sessions[id], tlsConfig, plainDialer{}) + }() + } + select { + case <-app.loggedOn: + case <-time.After(10 * time.Second): + t.Fatal("the initiator did not log on") + } + + began := time.Now() + stopped := make(chan struct{}) + go func() { + initiator.Stop() + close(stopped) + }() + select { + case <-stopped: + case <-time.After(10 * time.Second): + t.Fatal("Stop did not return") + } + // Answered, not timed out: well inside the 5 s LogoutTimeout. + require.Less(t, time.Since(began), 4*time.Second, "the Logout was not answered") + select { + case <-counterparty.loggedOut: + case <-time.After(5 * time.Second): + t.Fatal("the counterparty never received the Logout") + } + }) + } +} From 8b6db7b296f2fa1c8173ddb4354d51f8d9e71cea Mon Sep 17 00:00:00 2001 From: Josh Kelly Date: Thu, 5 Nov 2026 20:45:19 +0900 Subject: [PATCH 2/2] fix(initiator): bound the tls handshake and close failed connections The TLS handshake took no context and had no deadline, although SocketTimeout is documented as its timeout: v0.9.10 applies it to the TCP dial only. A stop during a stalled handshake therefore waited until the kernel gave up on the connection, and a counterparty that accepted the connection and never answered held the attempt forever, with no reconnect. A failed handshake, or a failed session start, also went back to reconnect without closing the connection it had dialed, leaking one socket per attempt. The handshake now runs under the dial context, so Stop cancels it, within SocketTimeout, or LogonTimeout when SocketTimeout is unset. Both failure paths close the transport before reconnecting. loadSocketTimeout reads the setting for the dialer and the handshake alike. Tests: Stop during a handshake the peer never answers, a stalled handshake that times out and reconnects, a failed handshake that closes its connection, and a slow handshake inside SocketTimeout that is not cut short by LogonTimeout. --- config/configuration.go | 2 +- dialer.go | 33 +++++--- initiator.go | 16 +++- initiator_test.go | 170 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 207 insertions(+), 14 deletions(-) diff --git a/config/configuration.go b/config/configuration.go index 6282bab1e..598e48d30 100644 --- a/config/configuration.go +++ b/config/configuration.go @@ -587,7 +587,7 @@ const ( // // Required: No // - // Default: 0 (no timeout) + // Default: 0 (the handshake is bounded by LogonTimeout) // // Valid Values: // - A valid go time.Duration diff --git a/dialer.go b/dialer.go index de4419ff8..55f345ffb 100644 --- a/dialer.go +++ b/dialer.go @@ -25,21 +25,30 @@ import ( "github.com/quickfixgo/quickfix/config" ) +// loadSocketTimeout reads SocketTimeout: a duration, else whole seconds. It +// is zero when the setting is absent. +func loadSocketTimeout(settings *SessionSettings) (time.Duration, error) { + if !settings.HasSetting(config.SocketTimeout) { + return 0, nil + } + timeout, err := settings.DurationSetting(config.SocketTimeout) + if err == nil { + return timeout, nil + } + timeoutInt, err := settings.IntSetting(config.SocketTimeout) + if err != nil { + return 0, err + } + return time.Duration(timeoutInt) * time.Second, nil +} + func loadDialerConfig(settings *SessionSettings) (dialer proxy.ContextDialer, err error) { stdDialer := &net.Dialer{} - if settings.HasSetting(config.SocketTimeout) { - timeout, err := settings.DurationSetting(config.SocketTimeout) - if err != nil { - timeoutInt, err := settings.IntSetting(config.SocketTimeout) - if err != nil { - return stdDialer, err - } - - stdDialer.Timeout = time.Duration(timeoutInt) * time.Second - } else { - stdDialer.Timeout = timeout - } + timeout, err := loadSocketTimeout(settings) + if err != nil { + return stdDialer, err } + stdDialer.Timeout = timeout dialer = stdDialer if !settings.HasSetting(config.ProxyType) { diff --git a/initiator.go b/initiator.go index fd54d7240..cef030602 100644 --- a/initiator.go +++ b/initiator.go @@ -156,6 +156,14 @@ func (i *Initiator) handleConnection(session *session, tlsConfig *tls.Config, di wg.Wait() }() + // SocketTimeout bounds the TLS handshake. Without it LogonTimeout does, + // so a counterparty that accepts the connection and never answers cannot + // hold a connection attempt open. + handshakeTimeout := session.LogonTimeout + if timeout, err := loadSocketTimeout(i.sessionSettings[session.sessionID]); err == nil && timeout > 0 { + handshakeTimeout = timeout + } + connectionAttempt := 0 for { @@ -198,9 +206,14 @@ func (i *Initiator) handleConnection(session *session, tlsConfig *tls.Config, di } tlsConfig.ServerName = serverName } + // The handshake ends with the dial context, so Stop cancels it. tlsConn := tls.Client(netConn, tlsConfig) - if err = tlsConn.Handshake(); err != nil { + handshakeCtx, cancelHandshake := context.WithTimeout(ctx, handshakeTimeout) + err = tlsConn.HandshakeContext(handshakeCtx) + cancelHandshake() + if err != nil { session.log.OnEventf("Failed handshake: %v", err) + closeTransport(netConn, session.log) goto reconnect } netConn = tlsConn @@ -210,6 +223,7 @@ func (i *Initiator) handleConnection(session *session, tlsConfig *tls.Config, di msgOut = make(chan []byte) if err := session.connect(msgIn, msgOut); err != nil { session.log.OnEventf("Failed to initiate: %v", err) + closeTransport(netConn, session.log) goto reconnect } diff --git a/initiator_test.go b/initiator_test.go index cb22dac8a..5a8596596 100644 --- a/initiator_test.go +++ b/initiator_test.go @@ -470,3 +470,173 @@ HeartBtInt=30 }) } } + +// silentTLSPeer accepts TCP connections and never answers the TLS +// handshake. Each accepted connection is handed to the test. +func silentTLSPeer(t *testing.T) (int, chan net.Conn) { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + accepted := make(chan net.Conn, 16) + t.Cleanup(func() { _ = listener.Close() }) + go func() { + for { + conn, err := listener.Accept() + if err != nil { + return + } + t.Cleanup(func() { _ = conn.Close() }) + accepted <- conn + } + }() + return listener.Addr().(*net.TCPAddr).Port, accepted +} + +// closedByPeer reports whether the other side closed conn within timeout: +// it drains what was sent and waits for the end of the stream. +func closedByPeer(conn net.Conn, timeout time.Duration) bool { + _ = conn.SetReadDeadline(time.Now().Add(timeout)) + _, err := io.Copy(io.Discard, conn) + return !errors.Is(err, os.ErrDeadlineExceeded) +} + +// startTLSInitiator runs one initiator session against port over TLS, the +// way Start would, and returns the initiator. +func startTLSInitiator(t *testing.T, compIDs string, port int, extra string, tlsConfig *tls.Config) *Initiator { + t.Helper() + settings, err := ParseSettings(strings.NewReader(fmt.Sprintf(`[DEFAULT] +ReconnectInterval=1 +%s +[SESSION] +BeginString=FIX.4.4 +SenderCompID=%s_INITIATOR +TargetCompID=%s_ACCEPTOR +SocketConnectHost=127.0.0.1 +SocketConnectPort=%d +HeartBtInt=30 +`, extra, compIDs, compIDs, port))) + require.NoError(t, err) + initiator, err := NewInitiator(logonRecorder{loggedOn: make(chan SessionID, 1), loggedOut: make(chan SessionID, 1)}, NewMemoryStoreFactory(), settings, NewNullLogFactory()) + require.NoError(t, err) + initiator.stopChan = make(chan interface{}) + for id := range initiator.sessionSettings { + initiator.wg.Add(1) + go func() { + defer initiator.wg.Done() + initiator.handleConnection(initiator.sessions[id], tlsConfig, plainDialer{}) + }() + } + return initiator +} + +// Stop does not wait on a TLS handshake the counterparty never answers, and +// closes that connection. +func TestInitiatorStopDuringAStalledTLSHandshake(t *testing.T) { + port, accepted := silentTLSPeer(t) + initiator := startTLSInitiator(t, "HANDSHAKE_STOP", port, "LogonTimeout=60", &tls.Config{InsecureSkipVerify: true}) //nolint:gosec // the peer never answers + var conn net.Conn + select { + case conn = <-accepted: + case <-time.After(5 * time.Second): + t.Fatal("the initiator did not connect") + } + + stopped := make(chan struct{}) + go func() { + initiator.Stop() + close(stopped) + }() + select { + case <-stopped: + case <-time.After(3 * time.Second): + _ = conn.Close() // let the stuck Stop end with the test + t.Fatal("Stop waited on the handshake") + } + require.True(t, closedByPeer(conn, 2*time.Second), "the handshake's connection was left open") +} + +// A TLS handshake the counterparty never answers fails after LogonTimeout, +// closes its connection, and the initiator connects again. +func TestInitiatorRetriesAStalledTLSHandshake(t *testing.T) { + port, accepted := silentTLSPeer(t) + initiator := startTLSInitiator(t, "HANDSHAKE_RETRY", port, "LogonTimeout=1", &tls.Config{InsecureSkipVerify: true}) //nolint:gosec // the peer never answers + t.Cleanup(initiator.Stop) + var first net.Conn + select { + case first = <-accepted: + case <-time.After(5 * time.Second): + t.Fatal("the initiator did not connect") + } + select { + case <-accepted: + case <-time.After(6 * time.Second): // LogonTimeout 1 s, then ReconnectInterval 1 s + t.Fatal("the initiator did not connect again after the stalled handshake") + } + require.True(t, closedByPeer(first, 2*time.Second), "the stalled handshake's connection was left open") +} + +// A failed TLS handshake closes the connection it was made on. +func TestInitiatorClosesAConnectionWhoseHandshakeFailed(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = listener.Close() }) + cert, _ := selfSignedCertificate(t) + failed := make(chan net.Conn, 1) + go func() { + raw, err := listener.Accept() + if err != nil { + return + } + t.Cleanup(func() { _ = raw.Close() }) + server := tls.Server(raw, &tls.Config{Certificates: []tls.Certificate{*cert}}) + _ = server.Handshake() // fails: the initiator does not trust this certificate + failed <- raw + }() + // An empty root pool trusts nothing. + initiator := startTLSInitiator(t, "HANDSHAKE_FAIL", listener.Addr().(*net.TCPAddr).Port, "LogonTimeout=60", &tls.Config{RootCAs: x509.NewCertPool(), MinVersion: tls.VersionTLS12}) + t.Cleanup(initiator.Stop) + var raw net.Conn + select { + case raw = <-failed: + case <-time.After(5 * time.Second): + t.Fatal("the handshake did not fail") + } + require.True(t, closedByPeer(raw, 2*time.Second), "the failed handshake's connection was left open") +} + +// SocketTimeout, when set, bounds the TLS handshake instead of LogonTimeout: +// a counterparty whose handshake takes longer than LogonTimeout but less +// than SocketTimeout still completes it, on the first connection. +func TestInitiatorHandshakeUsesSocketTimeout(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = listener.Close() }) + cert, roots := selfSignedCertificate(t) + accepts := make(chan error, 4) + go func() { + for { + raw, err := listener.Accept() + if err != nil { + return + } + t.Cleanup(func() { _ = raw.Close() }) + go func() { + time.Sleep(2 * time.Second) // a slow handshake + accepts <- tls.Server(raw, &tls.Config{Certificates: []tls.Certificate{*cert}}).Handshake() + }() + } + }() + initiator := startTLSInitiator(t, "HANDSHAKE_SLOW", listener.Addr().(*net.TCPAddr).Port, "LogonTimeout=1\nSocketTimeout=30s", &tls.Config{RootCAs: roots, MinVersion: tls.VersionTLS12}) + t.Cleanup(initiator.Stop) + select { + case err := <-accepts: + require.NoError(t, err, "the slow handshake did not complete") + case <-time.After(6 * time.Second): + t.Fatal("the handshake never finished") + } + select { + case err := <-accepts: + t.Fatalf("the initiator connected again after a handshake that succeeded: %v", err) + case <-time.After(2 * time.Second): + } +}