From 62c50e1685c28de390dd0d8f6eeb1f6af98e539b Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 26 Jul 2026 10:17:29 -0500 Subject: [PATCH 1/4] fix(ssh): release port-forward listener on transport death Local port forwards closed their listener only on ctx cancellation or idle timeout, never when the SSH transport died. On a network drop the accept loop stayed blocked on the dead client and the bound host port leaked, so reconnects failed with 'bind: address already in use'. Watch client.Wait() and cancel the forward context when the transport closes, releasing the listener and returning ErrTransportClosed so the port can be rebound on reconnect. Refs #759 --- pkg/ssh/forward.go | 36 ++++++++- pkg/ssh/forward_transport_test.go | 122 ++++++++++++++++++++++++++++++ 2 files changed, 155 insertions(+), 3 deletions(-) create mode 100644 pkg/ssh/forward_transport_test.go diff --git a/pkg/ssh/forward.go b/pkg/ssh/forward.go index f273e20e2..246c29712 100644 --- a/pkg/ssh/forward.go +++ b/pkg/ssh/forward.go @@ -18,6 +18,13 @@ import ( // should check for this error with errors.Is. var ErrIdleTimeout = errors.New("port forward idle timeout") +// ErrTransportClosed is returned when the forwarder shuts down because its +// underlying SSH transport was closed (e.g. a network drop that tripped the +// server keep-alive). Callers may treat it as a signal to reconnect. It exists +// so the local listener is released promptly instead of leaking the bound port +// until ctx cancellation. +var ErrTransportClosed = errors.New("ssh transport closed") + type ForwardingFunction func( net.Conn, *ssh.Client, @@ -114,6 +121,25 @@ func portForwarding( } }() + // Release the listener when the SSH transport dies so a reconnect can + // rebind the same local port instead of failing with + // "bind: address already in use". Without this the accept loop blocks + // indefinitely on a dead client and the bound port leaks (issue #759). + if client != nil { + transportClosed := make(chan struct{}) + go func() { + _ = client.Wait() + close(transportClosed) + }() + go func() { + select { + case <-done: + case <-transportClosed: + cancel(ErrTransportClosed) + } + }() + } + counter := newConnectionCounter(fwdCtx, exitAfterTimeout, func() { log.Infof( "Stopping port-forward on %s: idle for a while. "+ @@ -126,10 +152,14 @@ func portForwarding( // waiting for a new connection connection, err := listener.Accept() if err != nil { - // If shutdown was caused by the idle timeout, surface that - // typed error so callers can choose to treat it as a clean exit. - if cause := context.Cause(fwdCtx); errors.Is(cause, ErrIdleTimeout) { + // Surface the typed cause when the accept failure came from our + // own shutdown (idle timeout or transport death) so callers can + // distinguish a clean exit / reconnect signal from a real error. + switch cause := context.Cause(fwdCtx); { + case errors.Is(cause, ErrIdleTimeout): return ErrIdleTimeout + case errors.Is(cause, ErrTransportClosed): + return ErrTransportClosed } return err } diff --git a/pkg/ssh/forward_transport_test.go b/pkg/ssh/forward_transport_test.go new file mode 100644 index 000000000..a5a2948f7 --- /dev/null +++ b/pkg/ssh/forward_transport_test.go @@ -0,0 +1,122 @@ +package ssh + +import ( + "context" + "crypto/ed25519" + "errors" + "net" + "testing" + "time" + + "golang.org/x/crypto/ssh" +) + +// startTestSSHServer stands up a minimal SSH server on loopback and returns a +// connected client plus a kill func that severs the transport (simulating the +// #759 keep-alive teardown / network drop). Channels are rejected; the client +// only needs a live transport we can later close. +func startTestSSHServer(t *testing.T) (client *ssh.Client, kill func()) { + t.Helper() + + _, priv, err := ed25519.GenerateKey(nil) + if err != nil { + t.Fatalf("generate host key: %v", err) + } + signer, err := ssh.NewSignerFromKey(priv) + if err != nil { + t.Fatalf("signer: %v", err) + } + + srvCfg := &ssh.ServerConfig{NoClientAuth: true} + srvCfg.AddHostKey(signer) + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + + serverConnCh := make(chan net.Conn, 1) + go func() { + nConn, aerr := ln.Accept() + if aerr != nil { + return + } + serverConnCh <- nConn + sConn, chans, reqs, herr := ssh.NewServerConn(nConn, srvCfg) + if herr != nil { + return + } + go ssh.DiscardRequests(reqs) + go func() { + for nc := range chans { + _ = nc.Reject(ssh.Prohibited, "no channels in test") + } + }() + _ = sConn + }() + + cConn, err := ssh.Dial("tcp", ln.Addr().String(), &ssh.ClientConfig{ + User: "test", + HostKeyCallback: ssh.InsecureIgnoreHostKey(), + Timeout: 2 * time.Second, + }) + if err != nil { + t.Fatalf("dial: %v", err) + } + + serverConn := <-serverConnCh + kill = func() { + _ = serverConn.Close() + _ = ln.Close() + } + t.Cleanup(func() { _ = cConn.Close(); kill() }) + return cConn, kill +} + +// TestPortForwarding_ReleasesListenerOnTransportDeath is the regression test for +// devsy issue #759's secondary bug. When the SSH transport dies, portForwarding +// must release the local listener (returning ErrTransportClosed) so a reconnect +// can rebind the same host port instead of failing with +// "bind: address already in use". +func TestPortForwarding_ReleasesListenerOnTransportDeath(t *testing.T) { + client, kill := startTestSSHServer(t) + + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + addr := lis.Addr().String() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + done := make(chan error, 1) + go func() { + // exitAfterTimeout=0 disables the idle timer, so only transport death + // or ctx cancellation can close the listener. + done <- portForwarding(ctx, client, lis, addr, "tcp", "127.0.0.1:9", 0, forward) + }() + + // Let the accept loop start, then sever the transport like a network drop. + time.Sleep(100 * time.Millisecond) + kill() + + // portForwarding must return promptly with ErrTransportClosed, having + // released its listener. + select { + case err := <-done: + if !errors.Is(err, ErrTransportClosed) { + t.Fatalf("expected ErrTransportClosed, got %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("portForwarding did not return after transport death (listener leaked)") + } + + // The host port must be rebindable now — this is the reconnect path that + // previously failed with "bind: address already in use". + l2, err := net.Listen("tcp", addr) + if err != nil { + t.Fatalf("rebind on %s failed after transport death: %v", addr, err) + } + _ = l2.Close() +} From 051906a0309dd36e901c9bbd32a722433c918dc8 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 26 Jul 2026 10:19:42 -0500 Subject: [PATCH 2/4] test(ssh): fold transport-death test into forward_test.go, drop comments --- pkg/ssh/forward.go | 12 --- pkg/ssh/forward_test.go | 95 +++++++++++++++++++++++ pkg/ssh/forward_transport_test.go | 122 ------------------------------ 3 files changed, 95 insertions(+), 134 deletions(-) delete mode 100644 pkg/ssh/forward_transport_test.go diff --git a/pkg/ssh/forward.go b/pkg/ssh/forward.go index 246c29712..c9b9aafad 100644 --- a/pkg/ssh/forward.go +++ b/pkg/ssh/forward.go @@ -18,11 +18,6 @@ import ( // should check for this error with errors.Is. var ErrIdleTimeout = errors.New("port forward idle timeout") -// ErrTransportClosed is returned when the forwarder shuts down because its -// underlying SSH transport was closed (e.g. a network drop that tripped the -// server keep-alive). Callers may treat it as a signal to reconnect. It exists -// so the local listener is released promptly instead of leaking the bound port -// until ctx cancellation. var ErrTransportClosed = errors.New("ssh transport closed") type ForwardingFunction func( @@ -121,10 +116,6 @@ func portForwarding( } }() - // Release the listener when the SSH transport dies so a reconnect can - // rebind the same local port instead of failing with - // "bind: address already in use". Without this the accept loop blocks - // indefinitely on a dead client and the bound port leaks (issue #759). if client != nil { transportClosed := make(chan struct{}) go func() { @@ -152,9 +143,6 @@ func portForwarding( // waiting for a new connection connection, err := listener.Accept() if err != nil { - // Surface the typed cause when the accept failure came from our - // own shutdown (idle timeout or transport death) so callers can - // distinguish a clean exit / reconnect signal from a real error. switch cause := context.Cause(fwdCtx); { case errors.Is(cause, ErrIdleTimeout): return ErrIdleTimeout diff --git a/pkg/ssh/forward_test.go b/pkg/ssh/forward_test.go index 6411f36d7..609460f54 100644 --- a/pkg/ssh/forward_test.go +++ b/pkg/ssh/forward_test.go @@ -2,6 +2,7 @@ package ssh import ( "context" + "crypto/ed25519" "errors" "net" "testing" @@ -91,3 +92,97 @@ func TestPortForwarding_ParentCancelNotIdleTimeout(t *testing.T) { t.Fatal("timed out waiting for portForwarding to return") } } + +func startTestSSHServer(t *testing.T) (client *ssh.Client, kill func()) { + t.Helper() + + _, priv, err := ed25519.GenerateKey(nil) + if err != nil { + t.Fatalf("generate host key: %v", err) + } + signer, err := ssh.NewSignerFromKey(priv) + if err != nil { + t.Fatalf("signer: %v", err) + } + + srvCfg := &ssh.ServerConfig{NoClientAuth: true} + srvCfg.AddHostKey(signer) + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + + serverConnCh := make(chan net.Conn, 1) + go func() { + nConn, aerr := ln.Accept() + if aerr != nil { + return + } + serverConnCh <- nConn + sConn, chans, reqs, herr := ssh.NewServerConn(nConn, srvCfg) + if herr != nil { + return + } + go ssh.DiscardRequests(reqs) + go func() { + for nc := range chans { + _ = nc.Reject(ssh.Prohibited, "no channels in test") + } + }() + _ = sConn + }() + + cConn, err := ssh.Dial("tcp", ln.Addr().String(), &ssh.ClientConfig{ + User: "test", + HostKeyCallback: ssh.InsecureIgnoreHostKey(), + Timeout: 2 * time.Second, + }) + if err != nil { + t.Fatalf("dial: %v", err) + } + + serverConn := <-serverConnCh + kill = func() { + _ = serverConn.Close() + _ = ln.Close() + } + t.Cleanup(func() { _ = cConn.Close(); kill() }) + return cConn, kill +} + +func TestPortForwarding_ReleasesListenerOnTransportDeath(t *testing.T) { + client, kill := startTestSSHServer(t) + + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + addr := lis.Addr().String() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + done := make(chan error, 1) + go func() { + done <- portForwarding(ctx, client, lis, addr, "tcp", "127.0.0.1:9", 0, forward) + }() + + time.Sleep(100 * time.Millisecond) + kill() + + select { + case err := <-done: + if !errors.Is(err, ErrTransportClosed) { + t.Fatalf("expected ErrTransportClosed, got %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("portForwarding did not return after transport death (listener leaked)") + } + + l2, err := net.Listen("tcp", addr) + if err != nil { + t.Fatalf("rebind on %s failed after transport death: %v", addr, err) + } + _ = l2.Close() +} diff --git a/pkg/ssh/forward_transport_test.go b/pkg/ssh/forward_transport_test.go deleted file mode 100644 index a5a2948f7..000000000 --- a/pkg/ssh/forward_transport_test.go +++ /dev/null @@ -1,122 +0,0 @@ -package ssh - -import ( - "context" - "crypto/ed25519" - "errors" - "net" - "testing" - "time" - - "golang.org/x/crypto/ssh" -) - -// startTestSSHServer stands up a minimal SSH server on loopback and returns a -// connected client plus a kill func that severs the transport (simulating the -// #759 keep-alive teardown / network drop). Channels are rejected; the client -// only needs a live transport we can later close. -func startTestSSHServer(t *testing.T) (client *ssh.Client, kill func()) { - t.Helper() - - _, priv, err := ed25519.GenerateKey(nil) - if err != nil { - t.Fatalf("generate host key: %v", err) - } - signer, err := ssh.NewSignerFromKey(priv) - if err != nil { - t.Fatalf("signer: %v", err) - } - - srvCfg := &ssh.ServerConfig{NoClientAuth: true} - srvCfg.AddHostKey(signer) - - ln, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - t.Fatalf("listen: %v", err) - } - - serverConnCh := make(chan net.Conn, 1) - go func() { - nConn, aerr := ln.Accept() - if aerr != nil { - return - } - serverConnCh <- nConn - sConn, chans, reqs, herr := ssh.NewServerConn(nConn, srvCfg) - if herr != nil { - return - } - go ssh.DiscardRequests(reqs) - go func() { - for nc := range chans { - _ = nc.Reject(ssh.Prohibited, "no channels in test") - } - }() - _ = sConn - }() - - cConn, err := ssh.Dial("tcp", ln.Addr().String(), &ssh.ClientConfig{ - User: "test", - HostKeyCallback: ssh.InsecureIgnoreHostKey(), - Timeout: 2 * time.Second, - }) - if err != nil { - t.Fatalf("dial: %v", err) - } - - serverConn := <-serverConnCh - kill = func() { - _ = serverConn.Close() - _ = ln.Close() - } - t.Cleanup(func() { _ = cConn.Close(); kill() }) - return cConn, kill -} - -// TestPortForwarding_ReleasesListenerOnTransportDeath is the regression test for -// devsy issue #759's secondary bug. When the SSH transport dies, portForwarding -// must release the local listener (returning ErrTransportClosed) so a reconnect -// can rebind the same host port instead of failing with -// "bind: address already in use". -func TestPortForwarding_ReleasesListenerOnTransportDeath(t *testing.T) { - client, kill := startTestSSHServer(t) - - lis, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - t.Fatalf("listen: %v", err) - } - addr := lis.Addr().String() - - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - done := make(chan error, 1) - go func() { - // exitAfterTimeout=0 disables the idle timer, so only transport death - // or ctx cancellation can close the listener. - done <- portForwarding(ctx, client, lis, addr, "tcp", "127.0.0.1:9", 0, forward) - }() - - // Let the accept loop start, then sever the transport like a network drop. - time.Sleep(100 * time.Millisecond) - kill() - - // portForwarding must return promptly with ErrTransportClosed, having - // released its listener. - select { - case err := <-done: - if !errors.Is(err, ErrTransportClosed) { - t.Fatalf("expected ErrTransportClosed, got %v", err) - } - case <-time.After(3 * time.Second): - t.Fatal("portForwarding did not return after transport death (listener leaked)") - } - - // The host port must be rebindable now — this is the reconnect path that - // previously failed with "bind: address already in use". - l2, err := net.Listen("tcp", addr) - if err != nil { - t.Fatalf("rebind on %s failed after transport death: %v", addr, err) - } - _ = l2.Close() -} From 8504396c37902635a1941f4a7b1fc169d5bc2953 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 26 Jul 2026 10:23:21 -0500 Subject: [PATCH 3/4] test(ssh): pin host key and use t.Context to satisfy gosec/modernize --- pkg/ssh/forward_test.go | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/pkg/ssh/forward_test.go b/pkg/ssh/forward_test.go index 609460f54..5a56639e4 100644 --- a/pkg/ssh/forward_test.go +++ b/pkg/ssh/forward_test.go @@ -135,7 +135,7 @@ func startTestSSHServer(t *testing.T) (client *ssh.Client, kill func()) { cConn, err := ssh.Dial("tcp", ln.Addr().String(), &ssh.ClientConfig{ User: "test", - HostKeyCallback: ssh.InsecureIgnoreHostKey(), + HostKeyCallback: ssh.FixedHostKey(signer.PublicKey()), Timeout: 2 * time.Second, }) if err != nil { @@ -160,12 +160,9 @@ func TestPortForwarding_ReleasesListenerOnTransportDeath(t *testing.T) { } addr := lis.Addr().String() - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - done := make(chan error, 1) go func() { - done <- portForwarding(ctx, client, lis, addr, "tcp", "127.0.0.1:9", 0, forward) + done <- portForwarding(t.Context(), client, lis, addr, "tcp", "127.0.0.1:9", 0, forward) }() time.Sleep(100 * time.Millisecond) From 33b43026cf2aeec2a2fa276123d38eb344526619 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 26 Jul 2026 11:34:45 -0500 Subject: [PATCH 4/4] fix(ssh): log transport-close cause; close test listener on dial failure --- pkg/ssh/forward.go | 4 +++- pkg/ssh/forward_test.go | 1 + 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/pkg/ssh/forward.go b/pkg/ssh/forward.go index c9b9aafad..f87d3ce10 100644 --- a/pkg/ssh/forward.go +++ b/pkg/ssh/forward.go @@ -119,7 +119,9 @@ func portForwarding( if client != nil { transportClosed := make(chan struct{}) go func() { - _ = client.Wait() + if werr := client.Wait(); werr != nil { + log.Debugf("ssh transport closed on %s: %v", srcAddr, werr) + } close(transportClosed) }() go func() { diff --git a/pkg/ssh/forward_test.go b/pkg/ssh/forward_test.go index 5a56639e4..78152de83 100644 --- a/pkg/ssh/forward_test.go +++ b/pkg/ssh/forward_test.go @@ -112,6 +112,7 @@ func startTestSSHServer(t *testing.T) (client *ssh.Client, kill func()) { if err != nil { t.Fatalf("listen: %v", err) } + t.Cleanup(func() { _ = ln.Close() }) serverConnCh := make(chan net.Conn, 1) go func() {