Add teleport networking subprocess for port/agent/x11 forwarding

constants.go

@@ -842,13 +842,10 @@ const (
 	// command execution (exec and shells).
 	ExecSubCommand = "exec"
 
-	// LocalForwardSubCommand is the sub-command Teleport uses to re-exec itself
-	// for local port forwarding.
-	LocalForwardSubCommand = "forwardv2"
-
-	// RemoteForwardSubCommand is the sub-command Teleport uses to re-exec itself
-	// for remote port forwarding.
-	RemoteForwardSubCommand = "remoteforward"
+	// NetworkingSubCommand is the sub-command Teleport uses to re-exec itself
+	// for networking operations. e.g. local/remote port forwarding, agent forwarding,
+	// or x11 forwarding.
+	NetworkingSubCommand = "networking"
 
 	// CheckHomeDirSubCommand is the sub-command Teleport uses to re-exec itself
 	// to check if the user's home directory exists.

integration/integration_test.go

@@ -4744,7 +4744,7 @@ func testX11Forwarding(t *testing.T, suite *integrationTestSuite) {
 					}
 
 					// Create and run an exec command twice. When ControlPath is set, this will cause
-					// re-use of the connection and creation of two sessions within  the connection.
+					// re-use of the connection and creation of two sessions within the connection.
 					for i := 0; i < 2; i++ {
 						execCmd, err := helpers.ExternalSSHCommand(helpers.CommandOptions{
 							ForcePTY:      true,

lib/client/x11_session.go

@@ -29,6 +29,7 @@ import (
 	tracessh "github.com/gravitational/teleport/api/observability/tracing/ssh"
 	"github.com/gravitational/teleport/lib/sshutils"
 	"github.com/gravitational/teleport/lib/sshutils/x11"
+	"github.com/gravitational/teleport/lib/utils"
 )
 
 // handleX11Forwarding handles X11 channel requests for the given server session.
@@ -198,7 +199,7 @@ func (ns *NodeSession) serveX11Channels(ctx context.Context, sess *tracessh.Sess
 			}
 		}()
 
-		if err := x11.Forward(ctx, xconn, xchan); err != nil {
+		if err := utils.ProxyConn(ctx, xconn, xchan); err != nil {
 			log.WithError(err).Debug("Encountered error during X11 forwarding")
 		}
 	})

lib/srv/ctx.go

@@ -20,8 +20,6 @@ package srv
 
 import (
 	"context"
-	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -188,47 +186,6 @@ type Server interface {
 	TargetMetadata() apievents.ServerMetadata
 }
 
-// childProcessError is used to provide an underlying error
-// from a re-executed Teleport child process to its parent.
-type childProcessError struct {
-	Code     int    `json:"code"`
-	RawError []byte `json:"rawError"`
-}
-
-// writeChildError encodes the provided error
-// as json and writes it to w. Special care
-// is taken to preserve the error type by
-// including the error code and raw message
-// so that [DecodeChildError] will return
-// the matching error type and message.
-func writeChildError(w io.Writer, err error) {
-	if w == nil || err == nil {
-		return
-	}
-
-	data, jerr := json.Marshal(err)
-	if jerr != nil {
-		return
-	}
-
-	_ = json.NewEncoder(w).Encode(childProcessError{
-		Code:     trace.ErrorToCode(err),
-		RawError: data,
-	})
-}
-
-// DecodeChildError consumes the output from a child
-// process decoding it from its raw form back into
-// a concrete error.
-func DecodeChildError(r io.Reader) error {
-	var c childProcessError
-	if err := json.NewDecoder(r).Decode(&c); err != nil {
-		return nil
-	}
-
-	return trace.ReadError(c.Code, c.RawError)
-}
-
 // IdentityContext holds all identity information associated with the user
 // logged on the connection.
 type IdentityContext struct {
@@ -433,20 +390,6 @@ type ServerContext struct {
 	// by the server.
 	AllowFileCopying bool
 
-	// x11rdy{r,w} is used to signal from the child process to the
-	// parent process when X11 forwarding is set up.
-	x11rdyr *os.File
-	x11rdyw *os.File
-
-	// err{r,w} is used to propagate errors from the child process to the
-	// parent process so the parent can get more information about why the child
-	// process failed and act accordingly.
-	errr *os.File
-	errw *os.File
-
-	// x11Config holds the xauth and XServer listener config for this session.
-	x11Config *X11Config
-
 	// JoinOnly is set if the connection was created using a join-only principal and may only be used to join other sessions.
 	JoinOnly bool
 
@@ -598,24 +541,6 @@ func NewServerContext(ctx context.Context, parent *sshutils.ConnectionContext, s
 	child.AddCloser(child.killShellr)
 	child.AddCloser(child.killShellw)
 
-	// Create pipe used to get X11 forwarding ready signal from the child process.
-	child.x11rdyr, child.x11rdyw, err = os.Pipe()
-	if err != nil {
-		childErr := child.Close()
-		return nil, nil, trace.NewAggregate(err, childErr)
-	}
-	child.AddCloser(child.x11rdyr)
-	child.AddCloser(child.x11rdyw)
-
-	// Create pipe used to get errors from the child process.
-	child.errr, child.errw, err = os.Pipe()
-	if err != nil {
-		childErr := child.Close()
-		return nil, nil, trace.NewAggregate(err, childErr)
-	}
-	child.AddCloser(child.errr)
-	child.AddCloser(child.errw)
-
 	return ctx, child, nil
 }
 
@@ -829,26 +754,12 @@ func (c *ServerContext) CheckSFTPAllowed(registry *SessionRegistry) error {
 }
 
 // OpenXServerListener opens a new XServer unix listener.
-func (c *ServerContext) OpenXServerListener(x11Req x11.ForwardRequestPayload, displayOffset, maxDisplays int) error {
-	l, display, err := x11.OpenNewXServerListener(displayOffset, maxDisplays, x11Req.ScreenNumber)
+func (c *ServerContext) HandleX11Listener(l net.Listener, singleConnection bool) error {
+	display, err := x11.ParseDisplayFromUnixSocket(l.Addr().String())
 	if err != nil {
 		return trace.Wrap(err)
 	}
-
-	err = c.setX11Config(&X11Config{
-		XServerUnixSocket: l.Addr().String(),
-		XAuthEntry: x11.XAuthEntry{
-			Display: display,
-			Proto:   x11Req.AuthProtocol,
-			Cookie:  x11Req.AuthCookie,
-		},
-	})
-	if err != nil {
-		l.Close()
-		return trace.Wrap(err)
-	}
-
-	c.AddCloser(l)
+	c.Parent().SetEnv(x11.DisplayEnv, display.String())
 
 	// Prepare X11 channel request payload
 	originHost, originPort, err := net.SplitHostPort(c.ServerConn.LocalAddr().String())
@@ -868,26 +779,16 @@ func (c *ServerContext) OpenXServerListener(x11Req x11.ForwardRequestPayload, di
 		for {
 			xconn, err := l.Accept()
 			if err != nil {
-				// listener is closed
+				if !utils.IsOKNetworkError(err) {
+					c.Logger.WithError(err).Debug("Encountered error accepting XServer connection")
+				}
 				return
 			}
 
 			go func() {
 				defer xconn.Close()
 
-				// If the session has not signaled that X11 forwarding is
-				// fully set up yet, then ignore any incoming connections.
-				// The client's session hasn't been fully set up yet so this
-				// could potentially be a break-in attempt.
-				if ok, err := c.x11Ready(); err != nil {
-					c.Logger.WithError(err).Debug("Failed to get X11 ready status")
-					return
-				} else if !ok {
-					c.Logger.WithError(err).Debug("Rejecting X11 request, XServer Proxy is not ready")
-					return
-				}
-
-				xchan, sin, err := c.ServerConn.OpenChannel(sshutils.X11ChannelRequest, x11ChannelReqPayload)
+				xchan, sin, err := c.ServerConn.OpenChannel(x11.ChannelRequest, x11ChannelReqPayload)
 				if err != nil {
 					c.Logger.WithError(err).Debug("Failed to open a new X11 channel")
 					return
@@ -905,12 +806,12 @@ func (c *ServerContext) OpenXServerListener(x11Req x11.ForwardRequestPayload, di
 					}
 				}()
 
-				if err := x11.Forward(ctx, xconn, xchan); err != nil {
+				if err := utils.ProxyConn(ctx, xconn, xchan); err != nil {
 					c.Logger.WithError(err).Debug("Encountered error during X11 forwarding")
 				}
 			}()
 
-			if x11Req.SingleConnection {
+			if singleConnection {
 				l.Close()
 				return
 			}
@@ -920,52 +821,6 @@ func (c *ServerContext) OpenXServerListener(x11Req x11.ForwardRequestPayload, di
 	return nil
 }
 
-// getX11Config gets the x11 config for this server session.
-func (c *ServerContext) getX11Config() X11Config {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	if c.x11Config != nil {
-		return *c.x11Config
-	}
-	return X11Config{}
-}
-
-// setX11Config sets X11 config for the session, or returns an error if already set.
-func (c *ServerContext) setX11Config(cfg *X11Config) error {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	if c.x11Config != nil {
-		return trace.AlreadyExists("X11 forwarding is already set up for this session")
-	}
-
-	c.x11Config = cfg
-	return nil
-}
-
-// x11Ready returns whether the X11 unix listener is ready to accept connections.
-func (c *ServerContext) x11Ready() (bool, error) {
-	// Wait for child process to send signal (1 byte)
-	// or EOF if signal was already received.
-	_, err := io.ReadFull(c.x11rdyr, make([]byte, 1))
-	if errors.Is(err, io.EOF) {
-		return true, nil
-	} else if err != nil {
-		return false, trace.Wrap(err)
-	}
-
-	// signal received, close writer so future calls read EOF.
-	if err := c.x11rdyw.Close(); err != nil {
-		return false, trace.Wrap(err)
-	}
-	return true, nil
-}
-
-// GetChildError returns the error from the child process
-func (c *ServerContext) GetChildError() error {
-	return DecodeChildError(c.errr)
-}
-
 // takeClosers returns all resources that should be closed and sets the properties to null
 // we do this to avoid calling Close() under lock to avoid potential deadlocks
 func (c *ServerContext) takeClosers() []io.Closer {
@@ -1201,7 +1056,6 @@ func (c *ServerContext) ExecCommand() (*ExecCommand, error) {
 		PAMConfig:             pamConfig,
 		IsTestStub:            c.IsTestStub,
 		UaccMetadata:          *uaccMetadata,
-		X11Config:             c.getX11Config(),
 	}, nil
 }
 

lib/srv/ctx_test.go

@@ -19,14 +19,11 @@
 package srv
 
 import (
-	"bytes"
 	"context"
-	"os/user"
 	"testing"
 
 	"github.com/gogo/protobuf/proto"
 	"github.com/google/go-cmp/cmp"
-	"github.com/gravitational/trace"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/crypto/ssh"
 	"google.golang.org/protobuf/testing/protocmp"
@@ -39,19 +36,6 @@ import (
 	"github.com/gravitational/teleport/lib/sshutils"
 )
 
-// TestDecodeChildError ensures that child error message marshaling
-// and unmarshaling returns the original values.
-func TestDecodeChildError(t *testing.T) {
-	var buf bytes.Buffer
-	require.NoError(t, DecodeChildError(&buf))
-
-	targetErr := trace.NotFound(user.UnknownUserError("test").Error())
-
-	writeChildError(&buf, targetErr)
-
-	require.ErrorIs(t, DecodeChildError(&buf), targetErr)
-}
-
 func TestCheckSFTPAllowed(t *testing.T) {
 	srv := newMockServer(t)
 	ctx := newTestServerContext(t, srv, nil)

lib/srv/exec_linux_test.go

@@ -32,12 +32,7 @@ import (
 	"time"
 
 	"github.com/gravitational/trace"
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-
-	"github.com/gravitational/teleport"
-	"github.com/gravitational/teleport/lib/utils"
-	"github.com/gravitational/teleport/lib/utils/host"
 )
 
 func TestOSCommandPrep(t *testing.T) {
@@ -137,36 +132,6 @@ func TestConfigureCommand(t *testing.T) {
 	require.NotContains(t, cmd.Env, unexpectedKey+"="+unexpectedValue)
 }
 
-func TestRootConfigureCommand(t *testing.T) {
-	utils.RequireRoot(t)
-
-	login := utils.GenerateLocalUsername(t)
-	_, err := host.UserAdd(login, nil, "", "", "")
-	require.NoError(t, err)
-	t.Cleanup(func() {
-		_, err := host.UserDel(login)
-		require.NoError(t, err)
-	})
-
-	srv := newMockServer(t)
-	scx := newExecServerContext(t, srv)
-	scx.Identity.Login = login
-	scx.ExecType = teleport.TCPIPForwardRequest
-
-	u, err := user.Lookup(login)
-	require.NoError(t, err)
-	uid, err := strconv.ParseUint(u.Uid, 10, 32)
-	require.NoError(t, err)
-	gid, err := strconv.ParseUint(u.Gid, 10, 32)
-	require.NoError(t, err)
-
-	cmd, err := ConfigureCommand(scx)
-	require.NoError(t, err)
-	// Verify that the configured command will run as the expected user.
-	assert.Equal(t, uint32(uid), cmd.SysProcAttr.Credential.Uid)
-	assert.Equal(t, uint32(gid), cmd.SysProcAttr.Credential.Gid)
-}
-
 // TestContinue tests if the process hangs if a continue signal is not sent
 // and makes sure the process continues once it has been sent.
 func TestContinue(t *testing.T) {

lib/srv/forward/sshserver.go

@@ -1286,7 +1286,7 @@ func (s *Server) dispatch(ctx context.Context, ch ssh.Channel, req *ssh.Request,
 		return s.handleEnvs(ctx, ch, req, scx)
 	case sshutils.SubsystemRequest:
 		return s.handleSubsystem(ctx, ch, req, scx)
-	case sshutils.X11ForwardRequest:
+	case x11.ForwardRequest:
 		return s.handleX11Forward(ctx, ch, req, scx)
 	case sshutils.AgentForwardRequest:
 		// to maintain interoperability with OpenSSH, agent forwarding requests
@@ -1361,7 +1361,7 @@ func (s *Server) handleX11ChannelRequest(ctx context.Context, nch ssh.NewChannel
 	defer sch.Close()
 
 	// setup outbound X11 channel to client
-	cch, cin, err := s.sconn.OpenChannel(sshutils.X11ChannelRequest, nch.ExtraData())
+	cch, cin, err := s.sconn.OpenChannel(x11.ChannelRequest, nch.ExtraData())
 	if err != nil {
 		s.log.Errorf("X11 channel fwd failed: %v", err)
 		return
@@ -1386,7 +1386,7 @@ func (s *Server) handleX11ChannelRequest(ctx context.Context, nch ssh.NewChannel
 		}
 	}()
 
-	if err := x11.Forward(ctx, cch, sch); err != nil {
+	if err := utils.ProxyConn(ctx, cch, sch); err != nil {
 		s.log.WithError(err).Debug("Encountered error during x11 forwarding")
 	}
 }