[v14] Fix data race in file descriptors

integration/assist/command_test.go

@@ -323,7 +323,7 @@ func registerAndSetupMockSSHNode(t *testing.T, ctx context.Context, testDir stri
 	// right now because we need to get a valid certificate. The certificate
 	// needs proper principals, which implies knowing the node ID. This only
 	// happens after the node has joined.
-	var sshListenerFds []servicecfg.FileDescriptor
+	var sshListenerFds []*servicecfg.FileDescriptor
 	sshAddr := helpers.NewListenerOn(t, "localhost", service.ListenerNodeSSH, &sshListenerFds)
 
 	node := registerMockSSHNode(t, ctx, sshAddr, testDir, rc)

integration/helpers/instance.go

@@ -276,7 +276,7 @@ type TeleInstance struct {
 	// Log specifies the instance logger
 	Log utils.Logger
 	InstanceListeners
-	Fds []servicecfg.FileDescriptor
+	Fds []*servicecfg.FileDescriptor
 }
 
 // InstanceConfig is an instance configuration
@@ -298,7 +298,7 @@ type InstanceConfig struct {
 	// Ports is a collection of instance ports.
 	Listeners *InstanceListeners
 
-	Fds []servicecfg.FileDescriptor
+	Fds []*servicecfg.FileDescriptor
 }
 
 // NewInstance creates a new Teleport process instance.
@@ -1031,7 +1031,7 @@ type ProxyConfig struct {
 	// Disable ALPN routing
 	DisableALPNSNIListener bool
 	// FileDescriptors holds FDs to be injected into the Teleport process
-	FileDescriptors []servicecfg.FileDescriptor
+	FileDescriptors []*servicecfg.FileDescriptor
 }
 
 // StartProxy starts another Proxy Server and connects it to the cluster.

integration/helpers/ports.go

@@ -45,13 +45,13 @@ type InstanceListeners struct {
 // listener setup for a given test. InstanceListenerSetupFuncs are useful when
 // you need to have some distance between the test configuration and actually
 // executing the listener setup.
-type InstanceListenerSetupFunc func(*testing.T, *[]servicecfg.FileDescriptor) *InstanceListeners
+type InstanceListenerSetupFunc func(*testing.T, *[]*servicecfg.FileDescriptor) *InstanceListeners
 
 // StandardListenerSetupOn returns a InstanceListenerSetupFunc that will create
 // a new InstanceListeners configured with each service listening on its own
 // port, all bound to the supplied address
-func StandardListenerSetupOn(addr string) func(t *testing.T, fds *[]servicecfg.FileDescriptor) *InstanceListeners {
-	return func(t *testing.T, fds *[]servicecfg.FileDescriptor) *InstanceListeners {
+func StandardListenerSetupOn(addr string) func(t *testing.T, fds *[]*servicecfg.FileDescriptor) *InstanceListeners {
+	return func(t *testing.T, fds *[]*servicecfg.FileDescriptor) *InstanceListeners {
 		return &InstanceListeners{
 			Web:           NewListenerOn(t, addr, service.ListenerProxyWeb, fds),
 			SSH:           NewListenerOn(t, addr, service.ListenerNodeSSH, fds),
@@ -65,15 +65,15 @@ func StandardListenerSetupOn(addr string) func(t *testing.T, fds *[]servicecfg.F
 
 // StandardListenerSetup creates an InstanceListeners configures with each service
 // listening on its own port, all bound to the loopback address
-func StandardListenerSetup(t *testing.T, fds *[]servicecfg.FileDescriptor) *InstanceListeners {
+func StandardListenerSetup(t *testing.T, fds *[]*servicecfg.FileDescriptor) *InstanceListeners {
 	return StandardListenerSetupOn(Loopback)(t, fds)
 }
 
 // SingleProxyPortSetupOn creates a constructor function that will in turn generate an
 // InstanceConfig that allows proxying of multiple protocols over a single port when
 // invoked.
-func SingleProxyPortSetupOn(addr string) func(*testing.T, *[]servicecfg.FileDescriptor) *InstanceListeners {
-	return func(t *testing.T, fds *[]servicecfg.FileDescriptor) *InstanceListeners {
+func SingleProxyPortSetupOn(addr string) func(*testing.T, *[]*servicecfg.FileDescriptor) *InstanceListeners {
+	return func(t *testing.T, fds *[]*servicecfg.FileDescriptor) *InstanceListeners {
 		ssh := NewListenerOn(t, addr, service.ListenerProxyWeb, fds)
 		return &InstanceListeners{
 			Web:               ssh,
@@ -89,13 +89,13 @@ func SingleProxyPortSetupOn(addr string) func(*testing.T, *[]servicecfg.FileDesc
 
 // SingleProxyPortSetup generates an InstanceConfig that allows proxying of multiple protocols
 // over a single port.
-func SingleProxyPortSetup(t *testing.T, fds *[]servicecfg.FileDescriptor) *InstanceListeners {
+func SingleProxyPortSetup(t *testing.T, fds *[]*servicecfg.FileDescriptor) *InstanceListeners {
 	return SingleProxyPortSetupOn(Loopback)(t, fds)
 }
 
 // WebReverseTunnelMuxPortSetup generates a listener config using the same port for web and
 // tunnel, and independent ports for all other services.
-func WebReverseTunnelMuxPortSetup(t *testing.T, fds *[]servicecfg.FileDescriptor) *InstanceListeners {
+func WebReverseTunnelMuxPortSetup(t *testing.T, fds *[]*servicecfg.FileDescriptor) *InstanceListeners {
 	web := NewListener(t, service.ListenerProxyTunnelAndWeb, fds)
 	return &InstanceListeners{
 		Web:           web,
@@ -108,7 +108,7 @@ func WebReverseTunnelMuxPortSetup(t *testing.T, fds *[]servicecfg.FileDescriptor
 }
 
 // SeparatePostgresPortSetup generates a listener config with a defined port for Postgres
-func SeparatePostgresPortSetup(t *testing.T, fds *[]servicecfg.FileDescriptor) *InstanceListeners {
+func SeparatePostgresPortSetup(t *testing.T, fds *[]*servicecfg.FileDescriptor) *InstanceListeners {
 	return &InstanceListeners{
 		Web:           NewListener(t, service.ListenerProxyWeb, fds),
 		SSH:           NewListener(t, service.ListenerNodeSSH, fds),
@@ -121,7 +121,7 @@ func SeparatePostgresPortSetup(t *testing.T, fds *[]servicecfg.FileDescriptor) *
 }
 
 // SeparateMongoPortSetup generates a listener config with a defined port for MongoDB
-func SeparateMongoPortSetup(t *testing.T, fds *[]servicecfg.FileDescriptor) *InstanceListeners {
+func SeparateMongoPortSetup(t *testing.T, fds *[]*servicecfg.FileDescriptor) *InstanceListeners {
 	return &InstanceListeners{
 		Web:           NewListener(t, service.ListenerProxyWeb, fds),
 		SSH:           NewListener(t, service.ListenerNodeSSH, fds),
@@ -134,7 +134,7 @@ func SeparateMongoPortSetup(t *testing.T, fds *[]servicecfg.FileDescriptor) *Ins
 }
 
 // SeparateMongoAndPostgresPortSetup generates a listener config with a defined port for Postgres and Mongo
-func SeparateMongoAndPostgresPortSetup(t *testing.T, fds *[]servicecfg.FileDescriptor) *InstanceListeners {
+func SeparateMongoAndPostgresPortSetup(t *testing.T, fds *[]*servicecfg.FileDescriptor) *InstanceListeners {
 	return &InstanceListeners{
 		Web:           NewListener(t, service.ListenerProxyWeb, fds),
 		SSH:           NewListener(t, service.ListenerNodeSSH, fds),
@@ -182,7 +182,7 @@ func Port(t *testing.T, addr string) int {
 //
 // The resulting file descriptor is added to the `fds` slice, which can then be
 // given to a teleport instance on startup in order to suppl
-func NewListenerOn(t *testing.T, hostAddr string, ty service.ListenerType, fds *[]servicecfg.FileDescriptor) string {
+func NewListenerOn(t *testing.T, hostAddr string, ty service.ListenerType, fds *[]*servicecfg.FileDescriptor) string {
 	t.Helper()
 
 	l, err := net.Listen("tcp", net.JoinHostPort(hostAddr, "0"))
@@ -194,20 +194,20 @@ func NewListenerOn(t *testing.T, hostAddr string, ty service.ListenerType, fds *
 	// the original net.Listener still needs to be closed.
 	lf, err := l.(*net.TCPListener).File()
 	require.NoError(t, err)
+	fd := &servicecfg.FileDescriptor{
+		Type:    string(ty),
+		Address: addr,
+		File:    lf,
+	}
 
 	// If the file descriptor slice ends up being passed to a TeleportProcess
 	// that successfully starts, listeners will either get "imported" and used
 	// or discarded and closed, this is just an extra safety measure that closes
 	// the listener at the end of the test anyway (the finalizer would do that
 	// anyway, in principle).
-	t.Cleanup(func() { lf.Close() })
-
-	*fds = append(*fds, servicecfg.FileDescriptor{
-		Type:    string(ty),
-		Address: addr,
-		File:    lf,
-	})
+	t.Cleanup(func() { require.NoError(t, fd.Close()) })
 
+	*fds = append(*fds, fd)
 	return addr
 }
 
@@ -221,7 +221,7 @@ func NewListenerOn(t *testing.T, hostAddr string, ty service.ListenerType, fds *
 //
 // The resulting file descriptor is added to the `fds` slice, which can then be
 // given to a teleport instance on startup in order to suppl
-func NewListener(t *testing.T, ty service.ListenerType, fds *[]servicecfg.FileDescriptor) string {
+func NewListener(t *testing.T, ty service.ListenerType, fds *[]*servicecfg.FileDescriptor) string {
 	return NewListenerOn(t, Loopback, ty, fds)
 }
 
@@ -230,7 +230,7 @@ func NewListener(t *testing.T, ty service.ListenerType, fds *[]servicecfg.FileDe
 // This is usefully when Teleport service is created from config file where a port is allocated by OS.
 type DynamicServiceAddr struct {
 	// Descriptors ia a list of descriptors associated with listens.
-	Descriptors []servicecfg.FileDescriptor
+	Descriptors []*servicecfg.FileDescriptor
 	// WebAddr is a Teleport Proxy Web Address.
 	WebAddr string
 	// TunnelAddr is a Teleport Proxy Tunnel Address.
@@ -245,7 +245,7 @@ type DynamicServiceAddr struct {
 
 // NewDynamicServiceAddr creates an instance of DynamicServiceAddr.
 func NewDynamicServiceAddr(t *testing.T) *DynamicServiceAddr {
-	var fds []servicecfg.FileDescriptor
+	var fds []*servicecfg.FileDescriptor
 	webAddr := NewListener(t, service.ListenerProxyWeb, &fds)
 	tunnelAddr := NewListener(t, service.ListenerProxyTunnel, &fds)
 	authAddr := NewListener(t, service.ListenerAuth, &fds)

integration/integration_test.go

@@ -3185,7 +3185,7 @@ func testMultiplexingTrustedClusters(t *testing.T, suite *integrationTestSuite)
 	trustedClusters(t, suite, trustedClusterTest{multiplex: true})
 }
 
-func standardPortsOrMuxSetup(t *testing.T, mux bool, fds *[]servicecfg.FileDescriptor) *helpers.InstanceListeners {
+func standardPortsOrMuxSetup(t *testing.T, mux bool, fds *[]*servicecfg.FileDescriptor) *helpers.InstanceListeners {
 	if mux {
 		return helpers.WebReverseTunnelMuxPortSetup(t, fds)
 	}

lib/service/service.go

@@ -374,7 +374,7 @@ type TeleportProcess struct {
 	registeredListeners []registeredListener
 	// importedDescriptors is a list of imported file descriptors
 	// passed by the parent process
-	importedDescriptors []servicecfg.FileDescriptor
+	importedDescriptors []*servicecfg.FileDescriptor
 	// listenersClosed is a flag that indicates that the process should not open
 	// new listeners (for instance, because we're shutting down and we've already
 	// closed all the listeners)
@@ -653,7 +653,7 @@ type Process interface {
 	WaitForSignals(context.Context) error
 	// ExportFileDescriptors exports service listeners
 	// file descriptors used by the process.
-	ExportFileDescriptors() ([]servicecfg.FileDescriptor, error)
+	ExportFileDescriptors() ([]*servicecfg.FileDescriptor, error)
 	// Shutdown starts graceful shutdown of the process,
 	// blocks until all resources are freed and go-routines are
 	// shut down.

lib/service/servicecfg/config.go

@@ -21,6 +21,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/ghodss/yaml"
@@ -188,7 +189,7 @@ type Config struct {
 
 	// FileDescriptors is an optional list of file descriptors for the process
 	// to inherit and use for listeners, used for in-process updates.
-	FileDescriptors []FileDescriptor
+	FileDescriptors []*FileDescriptor
 
 	// PollingPeriod is set to override default internal polling periods
 	// of sync agents, used to speed up integration tests.
@@ -579,6 +580,8 @@ func ApplyDefaults(cfg *Config) {
 // FileDescriptor is a file descriptor associated
 // with a listener
 type FileDescriptor struct {
+	once sync.Once
+
 	// Type is a listener type, e.g. auth:ssh
 	Type string
 	// Address is an address of the listener, e.g. 127.0.0.1:3025
@@ -587,12 +590,22 @@ type FileDescriptor struct {
 	File *os.File
 }
 
+func (fd *FileDescriptor) Close() error {
+	var err error
+	fd.once.Do(func() {
+		err = fd.File.Close()
+	})
+	return trace.Wrap(err)
+}
+
 func (fd *FileDescriptor) ToListener() (net.Listener, error) {
 	listener, err := net.FileListener(fd.File)
 	if err != nil {
 		return nil, err
 	}
-	fd.File.Close()
+	if err := fd.Close(); err != nil {
+		return nil, trace.Wrap(err)
+	}
 	return listener, nil
 }
 

lib/service/signals.go

@@ -220,11 +220,11 @@ func (process *TeleportProcess) closeImportedDescriptors(prefix string) error {
 	defer process.Unlock()
 
 	var errors []error
-	openDescriptors := make([]servicecfg.FileDescriptor, 0, len(process.importedDescriptors))
+	openDescriptors := make([]*servicecfg.FileDescriptor, 0, len(process.importedDescriptors))
 	for _, d := range process.importedDescriptors {
 		if strings.HasPrefix(d.Type, prefix) {
 			process.log.Infof("Closing imported but unused descriptor %v %v.", d.Type, d.Address)
-			errors = append(errors, d.File.Close())
+			errors = append(errors, d.Close())
 		} else {
 			openDescriptors = append(openDescriptors, d)
 		}
@@ -353,16 +353,16 @@ func (process *TeleportProcess) stopListeners() error {
 }
 
 // ExportFileDescriptors exports file descriptors to be passed to child process
-func (process *TeleportProcess) ExportFileDescriptors() ([]servicecfg.FileDescriptor, error) {
-	var out []servicecfg.FileDescriptor
+func (process *TeleportProcess) ExportFileDescriptors() ([]*servicecfg.FileDescriptor, error) {
+	var out []*servicecfg.FileDescriptor
 	process.Lock()
 	defer process.Unlock()
 	for _, r := range process.registeredListeners {
 		file, err := utils.GetListenerFile(r.listener)
 		if err != nil {
 			return nil, trace.Wrap(err)
 		}
-		out = append(out, servicecfg.FileDescriptor{
+		out = append(out, &servicecfg.FileDescriptor{
 			File:    file,
 			Type:    string(r.typ),
 			Address: r.address,
@@ -372,7 +372,7 @@ func (process *TeleportProcess) ExportFileDescriptors() ([]servicecfg.FileDescri
 }
 
 // importFileDescriptors imports file descriptors from environment if there are any
-func importFileDescriptors(log logrus.FieldLogger) ([]servicecfg.FileDescriptor, error) {
+func importFileDescriptors(log logrus.FieldLogger) ([]*servicecfg.FileDescriptor, error) {
 	// These files may be passed in by the parent process
 	filesString := os.Getenv(teleportFilesEnvVar)
 	os.Unsetenv(teleportFilesEnvVar)
@@ -432,7 +432,7 @@ type fileDescriptor struct {
 }
 
 // filesToString serializes file descriptors as well as accompanying information (like socket host and port)
-func filesToString(files []servicecfg.FileDescriptor) (string, error) {
+func filesToString(files []*servicecfg.FileDescriptor) (string, error) {
 	out := make([]fileDescriptor, len(files))
 	for i, f := range files {
 		out[i] = fileDescriptor{
@@ -453,14 +453,14 @@ func filesToString(files []servicecfg.FileDescriptor) (string, error) {
 }
 
 // filesFromString de-serializes the file descriptors and turns them in the os.Files
-func filesFromString(in string) ([]servicecfg.FileDescriptor, error) {
+func filesFromString(in string) ([]*servicecfg.FileDescriptor, error) {
 	var out []fileDescriptor
 	if err := json.Unmarshal([]byte(in), &out); err != nil {
 		return nil, err
 	}
-	files := make([]servicecfg.FileDescriptor, len(out))
+	files := make([]*servicecfg.FileDescriptor, len(out))
 	for i, o := range out {
-		files[i] = servicecfg.FileDescriptor{
+		files[i] = &servicecfg.FileDescriptor{
 			File:    os.NewFile(uintptr(o.FileFD), o.FileName),
 			Address: o.Address,
 			Type:    o.Type,
@@ -496,7 +496,7 @@ func (process *TeleportProcess) forkChild() error {
 		return trace.Wrap(err)
 	}
 
-	listenerFiles = append(listenerFiles, servicecfg.FileDescriptor{
+	listenerFiles = append(listenerFiles, &servicecfg.FileDescriptor{
 		File:    writePipe,
 		Type:    signalPipeName,
 		Address: "127.0.0.1:0",

lib/tbot/testhelpers/srv.go

@@ -50,8 +50,8 @@ type DefaultBotConfigOpts struct {
 // slice, which should be passed as exported file descriptors to NewTeleport;
 // this is to ensure that we keep the listening socket open, to prevent other
 // processes from using the same port before we're done with it.
-func DefaultConfig(t *testing.T) (*config.FileConfig, []servicecfg.FileDescriptor) {
-	var fds []servicecfg.FileDescriptor
+func DefaultConfig(t *testing.T) (*config.FileConfig, []*servicecfg.FileDescriptor) {
+	var fds []*servicecfg.FileDescriptor
 
 	fc := &config.FileConfig{
 		Global: config.Global{
@@ -80,7 +80,7 @@ func DefaultConfig(t *testing.T) (*config.FileConfig, []servicecfg.FileDescripto
 }
 
 // MakeAndRunTestAuthServer creates an auth server useful for testing purposes.
-func MakeAndRunTestAuthServer(t *testing.T, log utils.Logger, fc *config.FileConfig, fds []servicecfg.FileDescriptor) (auth *service.TeleportProcess) {
+func MakeAndRunTestAuthServer(t *testing.T, log utils.Logger, fc *config.FileConfig, fds []*servicecfg.FileDescriptor) (auth *service.TeleportProcess) {
 	t.Helper()
 
 	var err error

tool/tctl/common/helpers_test.go

@@ -209,7 +209,7 @@ func mustWriteIdentityFile(t *testing.T, fc *config.FileConfig, username string)
 
 type testServerOptions struct {
 	fileConfig      *config.FileConfig
-	fileDescriptors []servicecfg.FileDescriptor
+	fileDescriptors []*servicecfg.FileDescriptor
 	fakeClock       clockwork.FakeClock
 }
 
@@ -221,7 +221,7 @@ func withFileConfig(fc *config.FileConfig) testServerOptionFunc {
 	}
 }
 
-func withFileDescriptors(fds []servicecfg.FileDescriptor) testServerOptionFunc {
+func withFileDescriptors(fds []*servicecfg.FileDescriptor) testServerOptionFunc {
 	return func(options *testServerOptions) {
 		options.fileDescriptors = fds
 	}