fix(ssh): harden reconnect and port-forward recovery

Author: acmore

infra/sidecar/entrypoint.sh

@@ -23,18 +23,20 @@ OKDEV_WORKSPACE_PATH="__OKDEV_WORKSPACE_PATH__"
 
 # Find the dev container's PID 1 by looking for a process with a different root filesystem.
 # In a shared PID namespace, our PID 1 is the pause/sandbox container.
-# We look for the first "sleep infinity" or the init process of the dev container.
+# We look for the first long-lived process in the dev container.
 DEV_PID=""
-for pid in $(ls /proc | grep -E '^[0-9]+$' | sort -n); do
+for pid in $(ls /proc 2>/dev/null | grep -E '^[0-9]+$' | sort -n); do
   [ "$pid" = "1" ] && continue
   [ "$pid" = "$$" ] && continue
-  # Skip if we can't read the process
-  [ -r "/proc/$pid/root" ] || continue
-  # Check if this process has a different root than us (different container)
-  if ! [ "/proc/$pid/root" -ef "/proc/self/root" ]; then
+  # Use 2>/dev/null to suppress error if process disappears while we're checking.
+  [ -r "/proc/$pid/root" ] 2>/dev/null || continue
+  if ! [ "/proc/$pid/root" -ef "/proc/self/root" ] 2>/dev/null; then
     # Found a process in a different mount namespace — likely the dev container
-    DEV_PID="$pid"
-    break
+    # Verify it still exists before we commit to it.
+    if [ -d "/proc/$pid" ]; then
+      DEV_PID="$pid"
+      break
+    fi
   fi
 done
 
@@ -43,14 +45,16 @@ if [ -z "$DEV_PID" ]; then
   exit 1
 fi
 
-# Ensure current GID exists in /etc/group inside the target to suppress
+# Ensure current GID exists in /etc/group inside BOTH the sidecar and the target to suppress
 # "groups: cannot find name for group ID ..." warnings from login shells.
 CURRENT_GID="$(id -g 2>/dev/null || true)"
 if [ -n "$CURRENT_GID" ]; then
+  grep -q ":${CURRENT_GID}:" /etc/group 2>/dev/null || \
+    echo "okdev:x:${CURRENT_GID}:" >> /etc/group 2>/dev/null || true
   nsenter --target "$DEV_PID" --mount -- sh -c "
     grep -q \":${CURRENT_GID}:\" /etc/group 2>/dev/null || \
     echo \"okdev:x:${CURRENT_GID}:\" >> /etc/group 2>/dev/null || true
-  "
+  " 2>/dev/null || true
 fi
 
 # If a remote command was requested, execute it inside the dev container.
@@ -105,6 +109,18 @@ sed -i \
   /usr/local/bin/nsenter-dev.sh
 chmod +x /usr/local/bin/nsenter-dev.sh
 
+# Harden sshd_config for long-lived idle sessions.
+# Server-side keepalive: probe every 30s, tolerate 10 misses (5min of dead connection).
+# This complements the client-side ServerAliveInterval and keeps intermediate
+# connections (kubectl port-forward, load balancers) alive with bidirectional traffic.
+if ! grep -q "ClientAliveInterval" /etc/ssh/sshd_config; then
+  cat >> /etc/ssh/sshd_config << 'SSHD_KEEPALIVE'
+ClientAliveInterval 30
+ClientAliveCountMax 10
+TCPKeepAlive yes
+SSHD_KEEPALIVE
+fi
+
 # Add ForceCommand to sshd_config dynamically
 if ! grep -q "ForceCommand" /etc/ssh/sshd_config; then
   echo "ForceCommand /usr/local/bin/nsenter-dev.sh" >> /etc/ssh/sshd_config

internal/cli/portforward_helper.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"log/slog"
 	"net"
 	"strconv"
 	"strings"
@@ -28,7 +29,14 @@ func startManagedPortForwardWithClient(k interface {
 	doneCh := make(chan struct{})
 	go func() {
 		defer close(doneCh)
-		errCh <- k.PortForward(ctx, namespace, pod, forwards, io.Discard, io.Discard)
+		slog.Debug("starting port-forward worker", "pod", pod, "forwards", forwards)
+		err := k.PortForward(ctx, namespace, pod, forwards, io.Discard, io.Discard)
+		if err != nil {
+			slog.Debug("port-forward worker exited with error", "pod", pod, "error", err)
+		} else {
+			slog.Debug("port-forward worker exited normally", "pod", pod)
+		}
+		errCh <- err
 	}()
 	cancelAndWait := func() {
 		cancel()
@@ -70,7 +78,14 @@ func startManagedPortForwardNoProbeWithClient(k interface {
 	doneCh := make(chan struct{})
 	go func() {
 		defer close(doneCh)
-		errCh <- k.PortForward(ctx, namespace, pod, forwards, io.Discard, io.Discard)
+		slog.Debug("starting port-forward worker (no-probe)", "pod", pod, "forwards", forwards)
+		err := k.PortForward(ctx, namespace, pod, forwards, io.Discard, io.Discard)
+		if err != nil {
+			slog.Debug("port-forward worker (no-probe) exited with error", "pod", pod, "error", err)
+		} else {
+			slog.Debug("port-forward worker (no-probe) exited normally", "pod", pod)
+		}
+		errCh <- err
 	}()
 	cancelAndWait := func() {
 		cancel()
@@ -122,3 +137,29 @@ func allPortsReachable(ports []int) bool {
 	}
 	return true
 }
+
+// startPortForwardKeepalive periodically dials the local port-forward endpoint
+// to create new SPDY streams on the underlying connection. Each TCP connection
+// triggers the port-forward handler to create a fresh stream pair (data+error),
+// which generates SYN/FIN frames visible to all intermediaries (load balancers,
+// API server proxies, WebSocket gateways) — keeping the connection alive even
+// when they don't count data on existing streams as "activity".
+func startPortForwardKeepalive(ctx context.Context, localPort int, interval time.Duration) {
+	go func() {
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", localPort), 2*time.Second)
+				if err != nil {
+					slog.Debug("port-forward keepalive dial failed", "port", localPort, "error", err)
+					continue
+				}
+				_ = conn.Close()
+			}
+		}
+	}()
+}

internal/cli/ports.go

@@ -57,7 +57,7 @@ func newPortsCmd(opts *Options) *cobra.Command {
 				return fmt.Errorf("wait for sshd ready: %w", err)
 			}
 			alias := sshHostAlias(sn)
-			changed, err := ensureSSHConfigEntry(alias, sn, ns, cfg.Spec.SSH.User, cfg.Spec.SSH.RemotePort, keyPath, cfgPath, cfg.Spec.Ports)
+			changed, err := ensureSSHConfigEntry(alias, sn, ns, cfg.Spec.SSH.User, cfg.Spec.SSH.RemotePort, keyPath, cfgPath, cfg.Spec.Ports, cfg.Spec.SSH)
 			if err != nil {
 				return fmt.Errorf("update ~/.ssh/config for managed forwards: %w", err)
 			}
@@ -72,7 +72,7 @@ func newPortsCmd(opts *Options) *cobra.Command {
 			if running {
 				_ = stopManagedSSHForward(alias)
 			}
-			if err := startManagedSSHForward(alias); err != nil {
+			if err := startManagedSSHForwardWithForwards(alias, cfg.Spec.Ports, cfg.Spec.SSH); err != nil {
 				return fmt.Errorf("start managed SSH forwards: %w", err)
 			}
 			if changed {

internal/cli/ssh.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log/slog"
 	"net"
 	"os"
 	"os/exec"
@@ -78,9 +79,12 @@ func newSSHCmd(opts *Options) *cobra.Command {
 			if err != nil {
 				return err
 			}
+			pfKeepAliveCtx, pfKeepAliveCancel := context.WithCancel(cmd.Context())
+			startPortForwardKeepalive(pfKeepAliveCtx, usedLocalPort, 30*time.Second)
 			var pfMu sync.Mutex
 			currentCancelPF := cancelPF
 			defer func() {
+				pfKeepAliveCancel()
 				pfMu.Lock()
 				defer pfMu.Unlock()
 				if currentCancelPF != nil {
@@ -90,7 +94,7 @@ func newSSHCmd(opts *Options) *cobra.Command {
 
 			sshHost := sshHostAlias(sn)
 			cfgPath, _ := config.ResolvePath(opts.ConfigPath)
-			if _, cfgErr := ensureSSHConfigEntry(sshHost, sn, ns, user, remotePort, keyPath, cfgPath, cfg.Spec.Ports); cfgErr != nil {
+			if _, cfgErr := ensureSSHConfigEntry(sshHost, sn, ns, user, remotePort, keyPath, cfgPath, cfg.Spec.Ports, cfg.Spec.SSH); cfgErr != nil {
 				fmt.Fprintf(cmd.ErrOrStderr(), "warning: failed to update ~/.ssh/config: %v\n", cfgErr)
 			}
 
@@ -100,6 +104,7 @@ func newSSHCmd(opts *Options) *cobra.Command {
 				RemotePort:        remotePort,
 				KeepAliveInterval: time.Duration(cfg.Spec.SSH.KeepAliveInterval) * time.Second,
 				KeepAliveTimeout:  time.Duration(cfg.Spec.SSH.KeepAliveTimeout) * time.Second,
+				KeepAliveCountMax: cfg.Spec.SSH.KeepAliveCountMax,
 			}
 			if noTmux {
 				tm.Env = map[string]string{"OKDEV_NO_TMUX": "1"}
@@ -123,11 +128,15 @@ func newSSHCmd(opts *Options) *cobra.Command {
 					currentCancelPF()
 					currentCancelPF = nil
 				}
+				pfKeepAliveCancel()
 				cancel, lp, err := startSSHPortForwardWithFallback(newKubeClient(opts), ns, podName(sn), localPort, remotePort)
 				if err != nil {
 					return "", 0, err
 				}
 				currentCancelPF = cancel
+				newCtx, newCancel := context.WithCancel(cmd.Context())
+				pfKeepAliveCancel = newCancel
+				startPortForwardKeepalive(newCtx, lp, 30*time.Second)
 				return "127.0.0.1", lp, nil
 			})
 			var lastRTTWarnNanos atomic.Int64
@@ -196,14 +205,23 @@ func newSSHCmd(opts *Options) *cobra.Command {
 				}
 				return nil
 			}
-			if err := tm.OpenShell(); err != nil {
-				if !tm.IsConnected() || isIgnorableProxyIOError(err) {
-					fmt.Fprintln(cmd.ErrOrStderr(), "Connection lost. Session ended.")
-					return nil
+			// Shell loop: reconnect automatically when the connection drops.
+			for {
+				err := tm.OpenShell()
+				if err == nil {
+					return nil // clean exit (user typed exit/logout)
+				}
+				if isIgnorableProxyIOError(err) || !tm.IsConnected() {
+					fmt.Fprintln(cmd.ErrOrStderr(), "\nConnection lost. Reconnecting...")
+					if !tm.WaitConnected(cmd.Context()) {
+						fmt.Fprintln(cmd.ErrOrStderr(), "Reconnect failed. Session ended.")
+						return nil
+					}
+					fmt.Fprintln(cmd.ErrOrStderr(), "Reconnected.")
+					continue
 				}
 				return fmt.Errorf("ssh shell failed: %w", err)
 			}
-			return nil
 		},
 	}
 
@@ -313,7 +331,7 @@ func sshHostAlias(sessionName string) string {
 	return "okdev-" + sessionName
 }
 
-func ensureSSHConfigEntry(hostAlias, sessionName, namespace, user string, remotePort int, keyPath, okdevConfigPath string, forwards []config.PortMapping) (bool, error) {
+func ensureSSHConfigEntry(hostAlias, sessionName, namespace, user string, remotePort int, keyPath, okdevConfigPath string, forwards []config.PortMapping, sshSpec config.SSHSpec) (bool, error) {
 	home, err := os.UserHomeDir()
 	if err != nil {
 		return false, err
@@ -341,15 +359,9 @@ func ensureSSHConfigEntry(hostAlias, sessionName, namespace, user string, remote
 		"  UserKnownHostsFile /dev/null",
 		"  ProxyCommand " + proxyCmd,
 	}
-	for _, p := range forwards {
-		if p.Local <= 0 || p.Remote <= 0 {
-			continue
-		}
-		blockLines = append(blockLines, fmt.Sprintf("  LocalForward %d 127.0.0.1:%d", p.Local, p.Remote))
-	}
 	blockLines = append(blockLines,
-		"  ServerAliveInterval 30",
-		"  ServerAliveCountMax 10",
+		fmt.Sprintf("  ServerAliveInterval %d", sshSpec.KeepAliveInterval),
+		fmt.Sprintf("  ServerAliveCountMax %d", sshSpec.KeepAliveCountMax),
 		"  TCPKeepAlive yes",
 		"  LogLevel ERROR",
 		end,
@@ -449,14 +461,21 @@ func newSSHProxyCmd(opts *Options) *cobra.Command {
 			if err != nil {
 				return err
 			}
+			pfKeepAliveCtx, pfKeepAliveCancel := context.WithCancel(context.Background())
+			startPortForwardKeepalive(pfKeepAliveCtx, usedLocalPort, 30*time.Second)
+			defer pfKeepAliveCancel()
 			if cancelPF != nil {
 				defer cancelPF()
 			}
+			slog.Debug("ssh-proxy dialing local port-forward", "port", usedLocalPort)
 			conn, err := waitDialLocal(usedLocalPort, 10*time.Second)
 			if err != nil {
+				slog.Debug("ssh-proxy dial failed", "error", err)
 				return err
 			}
 			defer conn.Close()
+			slog.Debug("ssh-proxy connection established", "localAddr", conn.LocalAddr(), "remoteAddr", conn.RemoteAddr())
+
 			var wg sync.WaitGroup
 			var copyErr error
 			var once sync.Once
@@ -465,13 +484,16 @@ func newSSHProxyCmd(opts *Options) *cobra.Command {
 				if err == nil || errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) || errors.Is(err, syscall.EPIPE) || isIgnorableProxyIOError(err) {
 					return
 				}
+				slog.Debug("ssh-proxy copy error", "error", err)
 				once.Do(func() { copyErr = err })
 			}
 			wg.Add(2)
 			go func() {
 				defer wg.Done()
+				slog.Debug("ssh-proxy starting copy: stdin -> conn")
 				_, err := io.Copy(conn, os.Stdin)
 				setErr(err)
+				slog.Debug("ssh-proxy finished copy: stdin -> conn")
 				select {
 				case <-done:
 				default:
@@ -480,12 +502,15 @@ func newSSHProxyCmd(opts *Options) *cobra.Command {
 			}()
 			go func() {
 				defer wg.Done()
+				slog.Debug("ssh-proxy starting copy: conn -> stdout")
 				_, err := io.Copy(os.Stdout, conn)
 				setErr(err)
+				slog.Debug("ssh-proxy finished copy: conn -> stdout")
 				close(done)
 				_ = conn.Close()
 			}()
 			<-done
+			slog.Debug("ssh-proxy session finished", "error", copyErr)
 			return copyErr
 		},
 	}
@@ -500,7 +525,8 @@ func isIgnorableProxyIOError(err error) bool {
 	msg := strings.ToLower(err.Error())
 	return strings.Contains(msg, "broken pipe") ||
 		strings.Contains(msg, "use of closed network connection") ||
-		strings.Contains(msg, "connection reset by peer")
+		strings.Contains(msg, "connection reset by peer") ||
+		strings.Contains(msg, "remote command exited without exit status")
 }
 
 func waitDialLocal(localPort int, timeout time.Duration) (net.Conn, error) {
@@ -533,7 +559,7 @@ func sshControlSocketPath(hostAlias string) (string, error) {
 	return filepath.Join(dir, hostAlias+".sock"), nil
 }
 
-func startManagedSSHForward(hostAlias string) error {
+func startManagedSSHForward(hostAlias string, sshSpec config.SSHSpec) error {
 	socketPath, err := sshControlSocketPath(hostAlias)
 	if err != nil {
 		return err
@@ -542,19 +568,38 @@ func startManagedSSHForward(hostAlias string) error {
 	if err := check.Run(); err == nil {
 		return nil
 	}
-	cmd := exec.Command(
+	return startManagedSSHForwardWithForwards(hostAlias, nil, sshSpec)
+}
+
+func startManagedSSHForwardWithForwards(hostAlias string, forwards []config.PortMapping, sshSpec config.SSHSpec) error {
+	socketPath, err := sshControlSocketPath(hostAlias)
+	if err != nil {
+		return err
+	}
+	check := exec.Command("ssh", "-S", socketPath, "-O", "check", hostAlias)
+	if err := check.Run(); err == nil {
+		return nil
+	}
+	args := []string{
 		"ssh",
 		"-fN",
 		"-M",
 		"-S", socketPath,
-		"-o", "ControlPersist=600",
+		"-o", "ControlPersist=3600",
 		"-o", "ExitOnForwardFailure=no",
-		"-o", "ServerAliveInterval=30",
-		"-o", "ServerAliveCountMax=10",
+		"-o", fmt.Sprintf("ServerAliveInterval=%d", sshSpec.KeepAliveInterval),
+		"-o", fmt.Sprintf("ServerAliveCountMax=%d", sshSpec.KeepAliveCountMax),
 		"-o", "TCPKeepAlive=yes",
 		"-o", "LogLevel=ERROR",
-		hostAlias,
-	)
+	}
+	for _, p := range forwards {
+		if p.Local <= 0 || p.Remote <= 0 {
+			continue
+		}
+		args = append(args, "-L", fmt.Sprintf("%d:127.0.0.1:%d", p.Local, p.Remote))
+	}
+	args = append(args, hostAlias)
+	cmd := exec.Command(args[0], args[1:]...)
 	if out, err := cmd.CombinedOutput(); err != nil {
 		return fmt.Errorf("start managed ssh forward: %w (%s)", err, strings.TrimSpace(string(out)))
 	}