diff --git a/lib/client/client.go b/lib/client/client.go index 1372cd7b2da..02fb03ffd61 100644 --- a/lib/client/client.go +++ b/lib/client/client.go @@ -931,14 +931,26 @@ func (c *NodeClient) remoteListenAndForward(ctx context.Context, ln net.Listener "remote_addr", remoteAddr, ) log.InfoContext(ctx, "Starting remote port forwarding") + defer log.InfoContext(ctx, "Shutting down remote port forwarding", "error", ctx.Err()) - for ctx.Err() == nil { + for { conn, err := acceptWithContext(ctx, ln) if err != nil { - if ctx.Err() == nil { + switch { + // Caller closed context to stop forwarding. For example the user + // ran "tsh ssh -N -R" then hit Ctrl-C. + case ctx.Err() != nil: + return + // The remote server closed the connection. For example, client_idle_timeout + // was hit. + case errors.Is(err, io.EOF): + return + // Accepting forwarded connection failed, but listener may still + // accept. For example, a transient network issue. + default: log.ErrorContext(ctx, "Remote port forwarding failed", "error", err) + continue } - continue } go func() { @@ -947,7 +959,6 @@ func (c *NodeClient) remoteListenAndForward(ctx context.Context, ln net.Listener } }() } - log.InfoContext(ctx, "Shutting down remote port forwarding", "error", ctx.Err()) } // GetRemoteTerminalSize fetches the terminal size of a given SSH session. diff --git a/lib/srv/regular/sshserver.go b/lib/srv/regular/sshserver.go index ce620f4c58e..b2f1ed9924a 100644 --- a/lib/srv/regular/sshserver.go +++ b/lib/srv/regular/sshserver.go @@ -2460,20 +2460,49 @@ func (s *Server) handleTCPIPForwardRequest(ctx context.Context, ccx *sshutils.Co if err != nil { return trace.Wrap(err) } - defer scx.Close() listener, err := s.listenTCPIP(ctx, scx, scx.SrcAddr) if err != nil { + if serr := scx.Close(); serr != nil { + s.logger.DebugContext(ctx, "Failed while cleaning up request", + "request_type", teleport.TCPIPForwardRequest, + "server_context_close_error", serr, + "error", err) + } return trace.Wrap(err) } // If the client didn't request a specific port, the chosen port needs to // be reported back. - srcHost, _, err := sshutils.SplitHostPort(scx.SrcAddr) + srcHost, srcPort, err := sshutils.SplitHostPort(scx.SrcAddr) if err != nil { + if lerr := listener.Close(); lerr != nil { + s.logger.DebugContext(ctx, "Failed while cleaning up request", + "request_type", teleport.TCPIPForwardRequest, + "listener_close_error", lerr, + "error", err) + } + if serr := scx.Close(); serr != nil { + s.logger.DebugContext(ctx, "Failed while cleaning up request", + "request_type", teleport.TCPIPForwardRequest, + "server_context_close_error", serr, + "error", err) + } return trace.Wrap(err) } _, listenPort, err := sshutils.SplitHostPort(listener.Addr().String()) if err != nil { + if lerr := listener.Close(); lerr != nil { + s.logger.DebugContext(ctx, "Failed while cleaning up request", + "request_type", teleport.TCPIPForwardRequest, + "listener_close_error", lerr, + "error", err) + } + if serr := scx.Close(); serr != nil { + s.logger.DebugContext(ctx, "Failed while cleaning up request", + "request_type", teleport.TCPIPForwardRequest, + "server_context_close_error", serr, + "error", err) + } return trace.Wrap(err) } scx.SrcAddr = sshutils.JoinHostPort(srcHost, listenPort) @@ -2483,6 +2512,7 @@ func (s *Server) handleTCPIPForwardRequest(ctx context.Context, ccx *sshutils.Co // spawn remote forwarding handler to multiplex connections to the forwarded port go func() { + defer scx.Close() stopEvent := scx.GetPortForwardEvent(events.PortForwardRemoteEvent, events.PortForwardStopCode, scx.SrcAddr) defer s.emitAuditEventWithLog(ctx, &stopEvent) @@ -2525,6 +2555,7 @@ func (s *Server) handleTCPIPForwardRequest(ctx context.Context, ccx *sshutils.Co logger.WarnContext(ctx, "failed to open channel", "error", err) continue } + ch = scx.TrackActivity(ch) go ssh.DiscardRequests(rch) go io.Copy(io.Discard, ch.Stderr()) go func() { @@ -2548,11 +2579,7 @@ func (s *Server) handleTCPIPForwardRequest(ctx context.Context, ccx *sshutils.Co // Report addr back to the client. if r.WantReply { var payload []byte - req, err := sshutils.ParseTCPIPForwardReq(r.Payload) - if err != nil { - return trace.Wrap(err) - } - if req.Port == 0 { + if srcPort == 0 { payload = ssh.Marshal(struct { Port uint32 }{Port: uint32(listener.Addr().(*net.TCPAddr).Port)}) diff --git a/lib/srv/regular/sshserver_test.go b/lib/srv/regular/sshserver_test.go index d702fd3cf8c..a8c36a0b8a7 100644 --- a/lib/srv/regular/sshserver_test.go +++ b/lib/srv/regular/sshserver_test.go @@ -19,13 +19,15 @@ package regular import ( + "bufio" + "bytes" "context" + "crypto/rand" "crypto/tls" "encoding/json" "errors" "fmt" "io" - "io/fs" "log/slog" "net" "net/http" @@ -91,6 +93,7 @@ import ( "github.com/gravitational/teleport/session/networking/x11" "github.com/gravitational/teleport/session/pam/pamcfg" "github.com/gravitational/teleport/session/reexec" + "github.com/gravitational/teleport/session/shell" ) // teleportTestUser is additional user used for tests @@ -368,10 +371,13 @@ func newCustomFixture(t testing.TB, mutateCfg func(*authtest.ServerConfig), sshO // responded to appropriately. Namely, it ensures that a response is sent if the client // requests a reply whether processing the request was successful or not. func TestTerminalSizeRequest(t *testing.T) { + t.Parallel() + f := newFixtureWithoutDiskBasedLogging(t) ctx := t.Context() t.Run("Invalid session", func(t *testing.T) { + t.Parallel() ok, resp, err := f.ssh.clt.SendRequest(ctx, teleport.TerminalSizeRequest, true, []byte("1234")) require.NoError(t, err) require.False(t, ok) @@ -379,6 +385,7 @@ func TestTerminalSizeRequest(t *testing.T) { }) t.Run("Active session", func(t *testing.T) { + t.Parallel() se, err := f.ssh.clt.NewSession(ctx) require.NoError(t, err) defer se.Close() @@ -435,6 +442,7 @@ func TestTerminalSizeRequest(t *testing.T) { // - and we give the race detector a chance to detect any possible race // conditions on this code path. func TestMultipleExecCommands(t *testing.T) { + t.Parallel() f := newFixtureWithoutDiskBasedLogging(t) ctx := t.Context() @@ -564,14 +572,17 @@ func TestSessionAuditLog(t *testing.T) { require.Equal(t, events.AgentForwardCode, agentForwardEvent.GetCode()) require.True(t, agentForwardEvent.Status.Success) - // Request x11 forwarding, event should be emitted immediately. - clientXAuthEntry, err := x11.NewFakeXAuthEntry(x11.Display{}) - require.NoError(t, err) - err = x11forward.RequestForwarding(ctx, se, clientXAuthEntry) - require.NoError(t, err) + // Only run this part of the test if xauth is available on the host. + if os.Getenv("TELEPORT_XAUTH_TEST") != "" { + // Request x11 forwarding, event should be emitted immediately. + clientXAuthEntry, err := x11.NewFakeXAuthEntry(x11.Display{}) + require.NoError(t, err) + err = x11forward.RequestForwarding(ctx, se, clientXAuthEntry) + require.NoError(t, err) - x11Event := <-emitter.C() - require.IsType(t, &apievents.X11Forward{}, x11Event, "expected X11Forward event but got event of type %T", x11Event) + x11Event := <-emitter.C() + require.IsType(t, &apievents.X11Forward{}, x11Event, "expected X11Forward event but got event of type %T", x11Event) + } // LOCAL PORT FORWARDING // Start up a test server that doesn't do any remote port forwarding @@ -844,6 +855,7 @@ func waitForBytes(ctx context.Context, ch <-chan []byte) ([]byte, error) { } func TestInactivityTimeout(t *testing.T) { + t.Parallel() const timeoutMessage = "You snooze, you lose." // Given @@ -879,7 +891,7 @@ func TestInactivityTimeout(t *testing.T) { return false } } - require.Eventually(t, sessionHasFinished, 6*time.Second, 100*time.Millisecond, + require.Eventually(t, sessionHasFinished, 6*time.Second, 10*time.Millisecond, "Timed out waiting for session to finish") // Expect that the idle timeout has been delivered via stderr @@ -889,6 +901,7 @@ func TestInactivityTimeout(t *testing.T) { } t.Run("Normal timeout", func(t *testing.T) { + t.Parallel() f := newCustomFixture(t, mutateCfg) // If all goes well, the client will be closed by the time cleanup happens, @@ -901,6 +914,7 @@ func TestInactivityTimeout(t *testing.T) { }) t.Run("Reset timeout on input", func(t *testing.T) { + t.Parallel() f := newCustomFixture(t, mutateCfg) // If all goes well, the client will be closed by the time cleanup happens, @@ -1015,6 +1029,16 @@ func TestLockInForce(t *testing.T) { _, err = newClient.NewSession(ctx) require.Error(t, err) require.Contains(t, err.Error(), lockInForceMsg) + t.Cleanup(func() { + // The client is expected to be closed by the lock monitor therefore + // expect an error on this second attempt. + // + // Wait before calling Close() to allow the server to close the + // connection. Wait here under the hood will just try and read from the + // connection until it gets an EOF. + require.Error(t, newClient.Wait()) + require.Error(t, newClient.Close()) + }) // Once the lock is lifted, new sessions should go through without error. require.NoError(t, f.testSrv.Auth().DeleteLock(ctx, "test-lock")) @@ -1052,10 +1076,11 @@ func setPortForwarding(t *testing.T, ctx context.Context, f *sshTestFixture, leg require.NoError(t, err) } -// TestDirectTCPIP ensures that the server can create a "direct-tcpip" -// channel to the target address. The "direct-tcpip" channel is what port -// forwarding is built upon. +// TestDirectTCPIP ensures that the server can create a "direct-tcpip" channel +// to the target address. The "direct-tcpip" channel is what port forwarding is +// built upon. func TestDirectTCPIP(t *testing.T) { + t.Parallel() ctx := t.Context() setup := func(t *testing.T) (*sshTestFixture, *httptest.Server, *url.URL) { @@ -1074,6 +1099,7 @@ func TestDirectTCPIP(t *testing.T) { } t.Run("Local forwarding is successful", func(t *testing.T) { + t.Parallel() f, ts, u := setup(t) defer ts.Close() @@ -1100,6 +1126,7 @@ func TestDirectTCPIP(t *testing.T) { }) t.Run("Local forwarding fails when access is denied", func(t *testing.T) { + t.Parallel() f, ts, u := setup(t) defer ts.Close() @@ -1128,6 +1155,7 @@ func TestDirectTCPIP(t *testing.T) { }) t.Run("Local forwarding fails when access is denied by legacy config", func(t *testing.T) { + t.Parallel() f, ts, u := setup(t) defer ts.Close() @@ -1156,6 +1184,7 @@ func TestDirectTCPIP(t *testing.T) { }) t.Run("SessionJoinPrincipal cannot use direct-tcpip", func(t *testing.T) { + t.Parallel() f, ts, u := setup(t) defer ts.Close() @@ -1217,9 +1246,9 @@ func TestTCPIPForward(t *testing.T) { } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { + t.Parallel() f := newFixtureWithoutDiskBasedLogging(t) setPortForwarding(t, t.Context(), f, tc.legacyAllow, tc.remoteAllow, tc.localAllow) - // create a new client connection to the node which will have its permissions // calculated with the updated rules. clientConn, err := apissh.Dial(t.Context(), "tcp", f.ssh.srvAddress, f.ssh.cltConfig) @@ -1237,12 +1266,17 @@ func TestTCPIPForward(t *testing.T) { require.NoError(t, err) } - // Start up a test server that uses the port forwarded listener. - ts := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Create and configure httptest.Server instead of calling + // httptest.NewUnstartedServer. Calling httptest.NewUnstartedServer + // creates an extra listener that is unneeded. + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { fmt.Fprintln(w, "hello, world") - })) + }) + ts := &httptest.Server{ + Listener: listener, + Config: &http.Server{Handler: handler}, + } t.Cleanup(ts.Close) - ts.Listener = listener ts.Start() // Dial the test server over the SSH connection. @@ -1265,6 +1299,7 @@ func TestTCPIPForward(t *testing.T) { } t.Run("SessionJoinPrincipal cannot use tcpip-forward", func(t *testing.T) { + t.Parallel() // Ensure that ssh client using SessionJoinPrincipal as Login, cannot // connect using "tcpip-forward". f := newFixtureWithoutDiskBasedLogging(t) @@ -1327,6 +1362,128 @@ func TestTCPIPForward(t *testing.T) { }) } +// TestRemotePortForwardNoSessionClientIdleTimeout verifies that traffic on +// a remote port forward without a session respects client_idle_timeout. +// It models `ssh -N -R : server01` behavior. +func TestRemotePortForwardNoSessionClientIdleTimeout(t *testing.T) { + t.Parallel() + + const idleTimeout = 5 * time.Second + + f := newCustomFixture(t, func(cfg *authtest.ServerConfig) { + networkCfg := types.DefaultClusterNetworkingConfig() + networkCfg.SetClientIdleTimeout(idleTimeout) + cfg.Auth.ClusterNetworkingConfig = networkCfg + }) + setPortForwarding(t, t.Context(), f, nil, types.NewBoolOption(true), nil) + + // Dial to the SSH server, perform a SSH handshake, then stop. Do not open a + // session channel (calls to NewSession, Run, Shell, or Start). This matches + // "ssh -N" behavior. + clientConn, err := apissh.Dial( + t.Context(), + "tcp", + f.ssh.srvAddress, + f.ssh.cltConfig, + ) + require.NoError(t, err) + t.Cleanup(func() { + clientConn.Close() + }) + + // Create a remote listener. Go provides Listen() on a clientConn that will + // create a listener on the remote end and then forward connections to the + // local client here. This matches "ssh -R" behavior. + listener, err := clientConn.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + client, err := net.Dial("tcp", listener.Addr().String()) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, client.Close()) + }) + + forwardConn, err := listener.Accept() + require.NoError(t, err) + t.Cleanup(func() { + require.ErrorIs(t, forwardConn.Close(), io.EOF) + }) + + // Read and write traffic on both sides of the connect at half the idle + // timeout interval over time advancing time beyond the idle timeout. Each + // exchange must reset the monitor deadline and keep the connection running. + for range 10 { + u := uuid.NewString() + + _, err := client.Write([]byte(u)) + require.NoError(t, err) + + request := make([]byte, len(u)) + _, err = io.ReadFull(forwardConn, request) + require.NoError(t, err) + require.Equal(t, u, string(request)) + + _, err = forwardConn.Write([]byte(u)) + require.NoError(t, err) + + response := make([]byte, len(u)) + _, err = io.ReadFull(client, response) + require.NoError(t, err) + require.Equal(t, u, string(response)) + + f.clock.Advance(idleTimeout / 2) + } + + // Forward the clock once more, this time past the idle timeout. The + // background monitor should now terminate the connection. + f.clock.Advance(idleTimeout + time.Second) + + // TODO(russjones): Update this test to use testing/synctest instead of a + // fake clock. + // + // This for-select loop is necessary because clockwork.Advance only fires + // timers that are registered when clockwork.Advance is called. The idle + // monitor in lib/srv/monitor.go rearms its timer asynchronously, leaving + // this race: + // + // For example, assume the idle timeout is five seconds: + // + // 1. At t=0, the monitor registers an idle timer to fire at t=5. + // 2. Client activity at t=2.5 means the connection should now expire at t=7.5. + // 3. At t=5, the original timer fires. The monitor observes the activity and + // calculates that 2.5 seconds remain before the connection is idle. + // 4. Before the monitor registers a new timer, the test advances time to + // t=7.5. No timer is registered to observe this advance. + // 5. The monitor then registers a 2.5-second timer relative to t=7.5, so it + // fires at t=10 instead of t=7.5. + // + // The second advance is therefore not observed by any idle timer, and the + // connection remains open until the monitor is scheduled and time is advanced + // again. + // + // The fix would be to use in-memory networking then synctest can wait for + // the monitor to become durably blocked before advancing virtual time. + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + timeout := time.NewTimer(1 * time.Second) + defer timeout.Stop() + waitCh := make(chan error, 1) + go func() { + waitCh <- clientConn.Wait() + }() + for { + select { + case err := <-waitCh: + require.ErrorIs(t, err, io.EOF) + return + case <-ticker.C: + f.clock.Advance(idleTimeout) + case <-timeout.C: + require.FailNow(t, "timed out waiting for idle timer to close connection") + } + } +} + func TestAdvertiseAddr(t *testing.T) { t.Parallel() f := newFixtureWithoutDiskBasedLogging(t) @@ -1501,8 +1658,9 @@ func TestOpenExecSessionSetsSession(t *testing.T) { // TestAgentForward tests agent forwarding via unix sockets func TestAgentForward(t *testing.T) { t.Parallel() - f := newFixtureWithoutDiskBasedLogging(t) + // Create and configure a cluster. + f := newFixtureWithoutDiskBasedLogging(t) ctx := t.Context() emitter := eventstest.NewChannelEmitter(32) f.ssh.srv.StreamEmitter = events.StreamerAndEmitter{ @@ -1519,9 +1677,11 @@ func TestAgentForward(t *testing.T) { _, err = f.testSrv.Auth().UpsertRole(ctx, role) require.NoError(t, err) + // Create new session, request agent forwarding on it, then close out agent. + // This is done to verify agent can run run across multiple sessions on a + // single connection. se, err := f.ssh.clt.NewSession(ctx) require.NoError(t, err) - t.Cleanup(func() { se.Close() }) err = sshagent.RequestAgentForwarding(ctx, se) require.NoError(t, err) @@ -1532,41 +1692,37 @@ func TestAgentForward(t *testing.T) { require.Equal(t, events.AgentForwardCode, agentForwardEvent.GetCode()) require.True(t, agentForwardEvent.Status.Success) - // prepare to send virtual "keyboard input" into the shell: - keyboard, err := se.StdinPipe() - require.NoError(t, err) - t.Cleanup(func() { keyboard.Close() }) - - // start interactive SSH session (new shell): - err = se.Shell(ctx) + err = se.Close() require.NoError(t, err) - // create a temp file to collect the shell output into: - tmpFile, err := os.CreateTemp(t.TempDir(), "teleport-agent-forward-test") - require.NoError(t, err) - tmpFile.Close() - - // type 'printenv SSH_AUTH_SOCK > /path/to/tmp/file' into the session (dumping the value of SSH_AUTH_STOCK into the temp file) - _, err = fmt.Fprintf(keyboard, "printenv %v >> %s\n\r", teleport.SSHAuthSock, tmpFile.Name()) + // Create another session (on same connection) extract path to agent. + se, err = f.ssh.clt.NewSession(ctx) require.NoError(t, err) - // wait for the output - var socketPath string - require.Eventually(t, func() bool { - output, err := os.ReadFile(tmpFile.Name()) - if err == nil && len(output) != 0 { - socketPath = strings.TrimSpace(string(output)) - return true - } - return false - }, 10*time.Second, 100*time.Millisecond, "failed to read socket path") + buf, err := se.Output(ctx, "printenv "+teleport.SSHAuthSock) + require.NoError(t, err) + socketPath := string(bytes.TrimSpace(buf)) - // try dialing the ssh agent socket: + // Close out this session. Calling Close() after Output() is expected to + // return EOF. See the following for more details: + // https://github.com/golang/go/issues/38115 + err = se.Close() + if err != nil && !errors.Is(err, io.EOF) { + require.NoError(t, err) + } + + // All sessions have been closed, verify agent can still be connected to. file, err := net.Dial("unix", socketPath) require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, file.Close()) + }) clientAgent := agent.NewClient(file) + _, err = clientAgent.List() + require.NoError(t, err) + sshConfig := apissh.ClientConfig{ User: f.user, PublicKeyAuth: apissh.PublicKeyAuthConfig{ @@ -1582,31 +1738,19 @@ func TestAgentForward(t *testing.T) { err = client.Close() require.NoError(t, err) - // make sure the socket persists after the session is closed. - // (agents are started from specific sessions, but apply to all - // sessions on the connection). - err = se.Close() - require.NoError(t, err) - - // Pause to allow closure to propagate. - time.Sleep(150 * time.Millisecond) - _, err = net.Dial("unix", socketPath) - require.NoError(t, err) - - // make sure the socket is gone after we closed the connection. Note that - // we now expect the client close to fail during the test cleanup, so we - // change the assertion accordingly + // Close the connection, after this errors are expected during cleanup, + // change assertion accordingly. require.NoError(t, f.ssh.clt.Close()) f.ssh.assertCltClose = require.Error - // clt must be nullified to prevent double-close during test cleanup - f.ssh.clt = nil - require.Eventually(t, func() bool { - _, err := clientAgent.List() - return err != nil - }, - 10*time.Second, 100*time.Millisecond, - "expected socket to be closed, still could dial") + // Set the read deadline to not wait for more than 10 seconds. This prevents + // this test blocking forever. Agent should no longer be available and + // return an error (other than deadline exceeded). + require.NoError(t, file.SetDeadline(time.Now().Add(10*time.Second))) + _, err = clientAgent.List() + require.Error(t, err) + require.NotContains(t, err.Error(), os.ErrDeadlineExceeded.Error(), + "timed out waiting for the server to close the agent connection") } // TestX11Forward tests x11 forwarding via unix sockets @@ -1671,9 +1815,14 @@ func TestX11Forward(t *testing.T) { // echoing XServer requests received back to the client. Returns the Display opened on the // session, which is set in $DISPLAY. func x11EchoSession(ctx context.Context, t *testing.T, clt *tracessh.Client) x11.Display { + // Create session but don't close it until the test is complete. This is key + // because the session must stay alive after this function returns for the + // proxy to be able to forward X11 requests. se, err := clt.NewSessionWithParams(ctx, nil) require.NoError(t, err) - t.Cleanup(func() { se.Close() }) + t.Cleanup(func() { + require.NoError(t, se.Close()) + }) // Create a fake client XServer listener which echos // back whatever it receives. @@ -1724,49 +1873,32 @@ func x11EchoSession(ctx context.Context, t *testing.T, clt *tracessh.Client) x11 err = x11forward.RequestForwarding(ctx, se, clientXAuthEntry) require.NoError(t, err) - // prepare to send virtual "keyboard input" into the shell: - keyboard, err := se.StdinPipe() + stdout, err := se.StdoutPipe() require.NoError(t, err) - // start interactive SSH session with x11 forwarding enabled (new shell): - err = se.Shell(ctx) - require.NoError(t, err) - - // create a temp file to collect the shell output into: - tmpFile, err := os.CreateTemp(os.TempDir(), "teleport-x11-forward-test") - require.NoError(t, err) - - // Allow non-root user to write to the temp file - err = tmpFile.Chmod(fs.FileMode(0o777)) + stdin, err := se.StdinPipe() require.NoError(t, err) t.Cleanup(func() { - os.Remove(tmpFile.Name()) + require.NoError(t, stdin.Close()) }) - // Reading the display may fail if the session is not fully initialized - // and the write to stdin is swallowed. - display := make(chan string, 1) - require.EventuallyWithT(t, func(t *assert.CollectT) { - // enter 'printenv DISPLAY > /path/to/tmp/file' into the session (dumping the value of DISPLAY into the temp file) - _, err = fmt.Fprintf(keyboard, "printenv %v > %s\n\r", x11.DisplayEnv, tmpFile.Name()) - require.NoError(t, err) + // Start an SSH exec command print the X11 display environment vaiable then + // block for the lifetime of the session. The SSH exec must block for the + // lifetime of the session to allow X11 requests to be forwarded. + // + // Blocking is done by cat reading from stdin and redirecting to /dev/null. + // Since nothing is written to stdin, this ends up blocking until the + // session/test is complete. + shell, err := shell.GetLoginShell(clt.User()) + require.NoError(t, err) + cmd := fmt.Sprintf("%v -c 'printenv %v\n'; exec cat >/dev/null", shell, x11.DisplayEnv) + err = se.Start(ctx, cmd) + require.NoError(t, err) - require.Eventually(t, func() bool { - output, err := os.ReadFile(tmpFile.Name()) - if err == nil && len(output) != 0 { - select { - case display <- strings.TrimSpace(string(output)): - default: - } - return true - } - return false - }, time.Second, 100*time.Millisecond, "failed to read display") - }, 10*time.Second, 1*time.Second) + display, err := bufio.NewReader(stdout).ReadString('\n') + require.NoError(t, err) - // Make a new connection to the XServer proxy, the client - // XServer should echo back anything written on it. - serverDisplay, err := x11.ParseDisplay(<-display) + serverDisplay, err := x11.ParseDisplay(strings.TrimSpace(display)) require.NoError(t, err) return serverDisplay @@ -1842,6 +1974,7 @@ func TestAllowedLabels(t *testing.T) { nil, ), ) + f.ssh.srv.dynamicLabels.Sync() tests := []struct { desc string @@ -1876,7 +2009,12 @@ func TestAllowedLabels(t *testing.T) { for _, tt := range tests { t.Run(tt.desc, func(t *testing.T) { - up, err := newUpack(t.Context(), f.testSrv, f.user, []string{f.user}, tt.inLabelMap) + t.Parallel() + + // Each case above requires updating the role. When run in parallel + // this becomes racy. Instead create a new user/role for each using + // rand.Text() so this test does not race. + up, err := newUpack(t.Context(), f.testSrv, rand.Text(), []string{f.user}, tt.inLabelMap) require.NoError(t, err) sshConfig := apissh.ClientConfig{ @@ -2517,7 +2655,7 @@ func TestLimiter(t *testing.T) { require.NoError(t, clt.Close()) require.ErrorIs(t, clt.Wait(), net.ErrClosed) - require.Eventually(t, getConns(t, limiter, "127.0.0.1", 1), time.Second*10, time.Millisecond*100) + require.Eventually(t, getConns(t, limiter, "127.0.0.1", 1), time.Second*10, time.Millisecond*10) // current connections = 1 clt, err = apissh.Dial(ctx, "tcp", srv.Addr(), config) @@ -2536,7 +2674,7 @@ func TestLimiter(t *testing.T) { require.NoError(t, clt.Close()) require.ErrorIs(t, clt.Wait(), net.ErrClosed) - require.Eventually(t, getConns(t, limiter, "127.0.0.1", 1), time.Second*10, time.Millisecond*100) + require.Eventually(t, getConns(t, limiter, "127.0.0.1", 1), time.Second*10, time.Millisecond*10) // current connections = 1 // requests rate should exceed now @@ -2841,6 +2979,7 @@ func requireNoErrors(t *testing.T, errsCh <-chan []error) { // TestParseSubsystemRequest verifies parseSubsystemRequest accepts the correct subsystems in depending on the runtime configuration. func TestParseSubsystemRequest(t *testing.T) { + t.Parallel() ctx := t.Context() // start a listener to accept connections; this will be needed for the proxy test to pass, otherwise nothing will be there to handle the call. @@ -3020,6 +3159,7 @@ func TestParseSubsystemRequest(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { + t.Parallel() subsystem := tt.name if tt.subsystemOverride != "" { subsystem = tt.subsystemOverride @@ -3307,6 +3447,7 @@ func TestHandlePuTTYWinadj(t *testing.T) { } func TestEventMetadata(t *testing.T) { + t.Parallel() ctx := t.Context() testServer, err := authtest.NewTestServer(authtest.ServerConfig{ Auth: authtest.AuthServerConfig{ @@ -3601,6 +3742,7 @@ func newSigner(t testing.TB, ctx context.Context, testServer *authtest.Server) s const maxPipeSize = 65536 + 1 func TestHostUserCreationProxy(t *testing.T) { + t.Parallel() f := newFixtureWithoutDiskBasedLogging(t) ctx := t.Context() @@ -3888,6 +4030,7 @@ func (f *fakeHostUsersBackend) SetHostUserDeletionGrace(grace time.Duration) { } func TestSessionParams(t *testing.T) { + t.Parallel() f := newFixtureWithoutDiskBasedLogging(t) ctx := t.Context() @@ -4001,6 +4144,7 @@ func TestSessionParams(t *testing.T) { } func TestServerInfo(t *testing.T) { + t.Parallel() scope := "/aa" f := newFixtureWithoutDiskBasedLogging(t, SetScope(scope)) require.Equal(t, f.ssh.srv.scope, scope) diff --git a/session/networking/x11/server.go b/session/networking/x11/server.go index 47206d9971c..31c85f46bba 100644 --- a/session/networking/x11/server.go +++ b/session/networking/x11/server.go @@ -60,10 +60,30 @@ func OpenNewXServerListener(displayOffset int, maxDisplay int, screen uint32) (n display := Display{DisplayNumber: displayNumber, ScreenNumber: int(screen)} if l, err := display.Listen(); err == nil { return l, display, nil - } else if !errors.Is(err, syscall.EADDRINUSE) && !errors.Is(err, syscall.EACCES) { + } else if tryNextDisplayError(err) { + // Skip to next display if the error is non-fatal. + continue + } else { return nil, Display{}, trace.Wrap(err) } } return nil, Display{}, trace.LimitExceeded("No more X11 sockets are available") } + +// tryNextDisplayError checks if the error is non-fatal and connecting to the +// next display should be attempted. The following cases are supported. +// +// * syscall.EADDRINUSE: Socket exists and something is already bound to it. +// This happens when a display is already being used by another X server. +// +// * syscall.EACCES: Permissions error. For example, socket may be owned by +// different user. +// +// * syscall.EEXIST: Socket already exists. Happens on macOS most often and in +// particular during heavy load on CI. +func tryNextDisplayError(err error) bool { + return errors.Is(err, syscall.EADDRINUSE) || + errors.Is(err, syscall.EACCES) || + errors.Is(err, syscall.EEXIST) +} diff --git a/session/reexec/reexec.go b/session/reexec/reexec.go index 109d730a9ad..8df968a4495 100644 --- a/session/reexec/reexec.go +++ b/session/reexec/reexec.go @@ -895,6 +895,7 @@ func RunNetworking() (code int, err error) { go func() { _, _ = terminatefd.Read(make([]byte, 1)) parentConn.Close() + cancel() }() // Alert the parent process that the child process is ready and listening for networking requests. @@ -979,6 +980,13 @@ func handleNetworkingRequest(ctx context.Context, conn *net.UnixConn, req networ conn.Write([]byte(trace.Wrap(err, "failed to write networking file to control conn").Error())) return nil } + + // Block for 30 seconds or until parent closes the request connection + // signaling it has a reference to the fd. This is to prevent the following + // race: child closes fd, but parent does not have reference to fd yet, and + // kernel comes and closes fd thinking it has 0 references. + <-ctx.Done() + return filePaths }