teleport/lib/service/service.go

/*
Copyright 2015-2021 Gravitational, Inc.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

// Package service implements teleport running service, takes care
// of initialization, cleanup and shutdown procedures
package service

import (
	"context"
	"crypto/rand"
	"crypto/tls"
	"encoding/hex"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"net/http"
	"net/http/httptest"
	"net/http/httputil"
	"net/http/pprof"
	"os"
	"path/filepath"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"golang.org/x/crypto/acme"
	"golang.org/x/crypto/acme/autocert"
	"golang.org/x/crypto/ssh"

	"github.com/gravitational/trace"

	"github.com/gravitational/teleport"
	"github.com/gravitational/teleport/api/client/proto"
	"github.com/gravitational/teleport/api/client/webclient"
	"github.com/gravitational/teleport/api/constants"
	apidefaults "github.com/gravitational/teleport/api/defaults"
	"github.com/gravitational/teleport/api/types"
	apievents "github.com/gravitational/teleport/api/types/events"
	"github.com/gravitational/teleport/lib/auth"
	"github.com/gravitational/teleport/lib/auth/native"
	"github.com/gravitational/teleport/lib/backend"
	"github.com/gravitational/teleport/lib/backend/dynamo"
	"github.com/gravitational/teleport/lib/backend/etcdbk"
	"github.com/gravitational/teleport/lib/backend/firestore"
	"github.com/gravitational/teleport/lib/backend/lite"
	"github.com/gravitational/teleport/lib/backend/memory"
	"github.com/gravitational/teleport/lib/bpf"
	"github.com/gravitational/teleport/lib/cache"
	"github.com/gravitational/teleport/lib/defaults"
	"github.com/gravitational/teleport/lib/events"
	"github.com/gravitational/teleport/lib/events/dynamoevents"
	"github.com/gravitational/teleport/lib/events/filesessions"
	"github.com/gravitational/teleport/lib/events/firestoreevents"
	"github.com/gravitational/teleport/lib/events/gcssessions"
	"github.com/gravitational/teleport/lib/events/s3sessions"
	kubeproxy "github.com/gravitational/teleport/lib/kube/proxy"
	"github.com/gravitational/teleport/lib/limiter"
	"github.com/gravitational/teleport/lib/modules"
	"github.com/gravitational/teleport/lib/multiplexer"
	"github.com/gravitational/teleport/lib/plugin"
	restricted "github.com/gravitational/teleport/lib/restrictedsession"
	"github.com/gravitational/teleport/lib/reversetunnel"
	"github.com/gravitational/teleport/lib/services"
	"github.com/gravitational/teleport/lib/session"
	"github.com/gravitational/teleport/lib/srv"
	"github.com/gravitational/teleport/lib/srv/app"
	"github.com/gravitational/teleport/lib/srv/db"
	"github.com/gravitational/teleport/lib/srv/regular"
	"github.com/gravitational/teleport/lib/system"
	"github.com/gravitational/teleport/lib/utils"
	"github.com/gravitational/teleport/lib/web"

	"github.com/gravitational/roundtrip"
	"github.com/jonboulle/clockwork"
	"github.com/pborman/uuid"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/sirupsen/logrus"
)

const (
	// AuthIdentityEvent is generated when the Auth Servers identity has been
	// initialized in the backend.
	AuthIdentityEvent = "AuthIdentity"

	// ProxyIdentityEvent is generated by the supervisor when the proxy's
	// identity has been registered with the Auth Server.
	ProxyIdentityEvent = "ProxyIdentity"

	// SSHIdentityEvent is generated when node's identity has been registered
	// with the Auth Server.
	SSHIdentityEvent = "SSHIdentity"

	// KubeIdentityEvent is generated by the supervisor when the kubernetes
	// service's identity has been registered with the Auth Server.
	KubeIdentityEvent = "KubeIdentity"

	// AppsIdentityEvent is generated when the identity of the application proxy
	// service has been registered with the Auth Server.
	AppsIdentityEvent = "AppsIdentity"

	// DatabasesIdentityEvent is generated when the identity of the database
	// proxy service has been registered with the auth server.
	DatabasesIdentityEvent = "DatabasesIdentity"

	// AuthTLSReady is generated when the Auth Server has initialized the
	// TLS Mutual Auth endpoint and is ready to start accepting connections.
	AuthTLSReady = "AuthTLSReady"

	// ProxyWebServerReady is generated when the proxy has initialized the web
	// server and is ready to start accepting connections.
	ProxyWebServerReady = "ProxyWebServerReady"

	// ProxyReverseTunnelReady is generated when the proxy has initialized the
	// reverse tunnel server and is ready to start accepting connections.
	ProxyReverseTunnelReady = "ProxyReverseTunnelReady"

	// DebugAppReady is generated when the debugging application has been started
	// and is ready to serve requests.
	DebugAppReady = "DebugAppReady"

	// ProxyAgentPoolReady is generated when the proxy has initialized the
	// remote cluster watcher (to spawn reverse tunnels) and is ready to start
	// accepting connections.
	ProxyAgentPoolReady = "ProxyAgentPoolReady"

	// ProxySSHReady is generated when the proxy has initialized a SSH server
	// and is ready to start accepting connections.
	ProxySSHReady = "ProxySSHReady"

	// NodeSSHReady is generated when the Teleport node has initialized a SSH server
	// and is ready to start accepting SSH connections.
	NodeSSHReady = "NodeReady"

	// KubernetesReady is generated when the kubernetes service has been initialized.
	KubernetesReady = "KubernetesReady"

	// AppsReady is generated when the Teleport app proxy service is ready to
	// start accepting connections.
	AppsReady = "AppsReady"

	// DatabasesReady is generated when the Teleport database proxy service
	// is ready to start accepting connections.
	DatabasesReady = "DatabasesReady"

	// TeleportExitEvent is generated when the Teleport process begins closing
	// all listening sockets and exiting.
	TeleportExitEvent = "TeleportExit"

	// TeleportReloadEvent is generated to trigger in-process teleport
	// service reload - all servers and clients will be re-created
	// in a graceful way.
	TeleportReloadEvent = "TeleportReload"

	// TeleportPhaseChangeEvent is generated to indidate that teleport
	// CA rotation phase has been updated, used in tests
	TeleportPhaseChangeEvent = "TeleportPhaseChange"

	// TeleportReadyEvent is generated to signal that all teleport
	// internal components have started successfully.
	TeleportReadyEvent = "TeleportReady"

	// ServiceExitedWithErrorEvent is emitted whenever a service
	// has exited with an error, the payload includes the error
	ServiceExitedWithErrorEvent = "ServiceExitedWithError"

	// TeleportDegradedEvent is emitted whenever a service is operating in a
	// degraded manner.
	TeleportDegradedEvent = "TeleportDegraded"

	// TeleportOKEvent is emitted whenever a service is operating normally.
	TeleportOKEvent = "TeleportOKEvent"
)

// RoleConfig is a configuration for a server role (either proxy or node)
type RoleConfig struct {
	DataDir     string
	HostUUID    string
	HostName    string
	AuthServers []utils.NetAddr
	Auth        AuthConfig
	Console     io.Writer
}

// Connector has all resources process needs to connect to other parts of the
// cluster: client and identity.
type Connector struct {
	// ClientIdentity is the identity to be used in internal cluster
	// clients to the auth service.
	ClientIdentity *auth.Identity

	// ServerIdentity is the identity to be used in servers - serving SSH
	// and x509 certificates to clients.
	ServerIdentity *auth.Identity

	// Client is authenticated client with credentials from ClientIdenity.
	Client *auth.Client
}

// TunnelProxy if non-empty, indicates that the client is connected to the Auth Server
// through the reverse SSH tunnel proxy
func (c *Connector) TunnelProxy() string {
	if c.Client == nil || c.Client.Dialer() == nil {
		return ""
	}
	tun, ok := c.Client.Dialer().(*reversetunnel.TunnelAuthDialer)
	if !ok {
		return ""
	}
	return tun.ProxyAddr
}

// UseTunnel indicates if the client is connected directly to the Auth Server
// (false) or through the proxy (true).
func (c *Connector) UseTunnel() bool {
	return c.TunnelProxy() != ""
}

// Close closes resources associated with connector
func (c *Connector) Close() error {
	if c.Client != nil {
		return c.Client.Close()
	}
	return nil
}

// TeleportProcess structure holds the state of the Teleport daemon, controlling
// execution and configuration of the teleport services: ssh, auth and proxy.
type TeleportProcess struct {
	Clock clockwork.Clock
	sync.Mutex
	Supervisor
	Config *Config

	// PluginsRegistry handles plugin registrations with Teleport services
	PluginRegistry plugin.Registry

	// localAuth has local auth server listed in case if this process
	// has started with auth server role enabled
	localAuth *auth.Server
	// backend is the process' backend
	backend backend.Backend
	// auditLog is the initialized audit log
	auditLog events.IAuditLog

	// identities of this process (credentials to auth sever, basically)
	Identities map[types.SystemRole]*auth.Identity

	// connectors is a list of connected clients and their identities
	connectors map[types.SystemRole]*Connector

	// registeredListeners keeps track of all listeners created by the process
	// used to pass listeners to child processes during live reload
	registeredListeners []registeredListener
	// importedDescriptors is a list of imported file descriptors
	// passed by the parent process
	importedDescriptors []FileDescriptor

	// forkedPIDs is a collection of a teleport processes forked
	// during restart used to collect their status in case if the
	// child process crashed.
	forkedPIDs []int

	// storage is a server local storage
	storage *auth.ProcessStorage

	// id is a process id - used to identify different processes
	// during in-process reloads.
	id string

	// log is a process-local log entry.
	log logrus.FieldLogger

	// keyPairs holds private/public key pairs used
	// to get signed host certificates from auth server
	keyPairs map[keyPairKey]KeyPair
	// keyMutex is a mutex to serialize key generation
	keyMutex sync.Mutex

	// reporter is used to report some in memory stats
	reporter *backend.Reporter

	// appDependCh is used by application service in single process mode to block
	// until auth and reverse tunnel servers are ready.
	appDependCh chan Event

	// clusterFeatures contain flags for supported and unsupported features.
	clusterFeatures proto.Features
}

type keyPairKey struct {
	role   types.SystemRole
	reason string
}

// processIndex is an internal process index
// to help differentiate between two different teleport processes
// during in-process reload.
var processID int32

func nextProcessID() int32 {
	return atomic.AddInt32(&processID, 1)
}

// GetAuthServer returns the process' auth server
func (process *TeleportProcess) GetAuthServer() *auth.Server {
	return process.localAuth
}

// GetAuditLog returns the process' audit log
func (process *TeleportProcess) GetAuditLog() events.IAuditLog {
	return process.auditLog
}

// GetBackend returns the process' backend
func (process *TeleportProcess) GetBackend() backend.Backend {
	return process.backend
}

func (process *TeleportProcess) findStaticIdentity(id auth.IdentityID) (*auth.Identity, error) {
	for i := range process.Config.Identities {
		identity := process.Config.Identities[i]
		if identity.ID.Equals(id) {
			return identity, nil
		}
	}
	return nil, trace.NotFound("identity %v not found", &id)
}

// getConnectors returns a copy of the identities registered for auth server
func (process *TeleportProcess) getConnectors() []*Connector {
	process.Lock()
	defer process.Unlock()

	out := make([]*Connector, 0, len(process.connectors))
	for role := range process.connectors {
		out = append(out, process.connectors[role])
	}
	return out
}

// addConnector adds connector to registered connectors list,
// it will overwrite the connector for the same role
func (process *TeleportProcess) addConnector(connector *Connector) {
	process.Lock()
	defer process.Unlock()

	process.connectors[connector.ClientIdentity.ID.Role] = connector
}

func (process *TeleportProcess) setClusterFeatures(features *proto.Features) {
	process.Lock()
	defer process.Unlock()

	if features != nil {
		process.clusterFeatures = *features
	}
}

func (process *TeleportProcess) getClusterFeatures() proto.Features {
	process.Lock()
	defer process.Unlock()

	return process.clusterFeatures
}

// GetIdentity returns the process identity (credentials to the auth server) for a given
// teleport Role. A teleport process can have any combination of 3 roles: auth, node, proxy
// and they have their own identities
func (process *TeleportProcess) GetIdentity(role types.SystemRole) (i *auth.Identity, err error) {
	var found bool

	process.Lock()
	defer process.Unlock()

	i, found = process.Identities[role]
	if found {
		return i, nil
	}
	i, err = process.storage.ReadIdentity(auth.IdentityCurrent, role)
	id := auth.IdentityID{
		Role:     role,
		HostUUID: process.Config.HostUUID,
		NodeName: process.Config.Hostname,
	}
	if err != nil {
		if !trace.IsNotFound(err) {
			return nil, trace.Wrap(err)
		}
		if role == types.RoleAdmin {
			// for admin identity use local auth server
			// because admin identity is requested by auth server
			// itself
			principals, dnsNames, err := process.getAdditionalPrincipals(role)
			if err != nil {
				return nil, trace.Wrap(err)
			}
			i, err = auth.GenerateIdentity(process.localAuth, id, principals, dnsNames)
			if err != nil {
				return nil, trace.Wrap(err)
			}
		} else {
			// try to locate static identity provided in the file
			i, err = process.findStaticIdentity(id)
			if err != nil {
				return nil, trace.Wrap(err)
			}
			process.log.Infof("Found static identity %v in the config file, writing to disk.", &id)
			if err = process.storage.WriteIdentity(auth.IdentityCurrent, *i); err != nil {
				return nil, trace.Wrap(err)
			}
		}
	}
	process.Identities[role] = i
	return i, nil
}

// Process is a interface for processes
type Process interface {
	// Closer closes all resources used by the process
	io.Closer
	// Start starts the process in a non-blocking way
	Start() error
	// WaitForSignals waits for and handles system process signals.
	WaitForSignals(context.Context) error
	// ExportFileDescriptors exports service listeners
	// file descriptors used by the process.
	ExportFileDescriptors() ([]FileDescriptor, error)
	// Shutdown starts graceful shutdown of the process,
	// blocks until all resources are freed and go-routines are
	// shut down.
	Shutdown(context.Context)
	// WaitForEvent waits for event to occur, sends event to the channel,
	// this is a non-blocking function.
	WaitForEvent(ctx context.Context, name string, eventC chan Event)
	// WaitWithContext waits for the service to stop. This is a blocking
	// function.
	WaitWithContext(ctx context.Context)
}

// NewProcess is a function that creates new teleport from config
type NewProcess func(cfg *Config) (Process, error)

func newTeleportProcess(cfg *Config) (Process, error) {
	return NewTeleport(cfg)
}

// Run starts teleport processes, waits for signals
// and handles internal process reloads.
func Run(ctx context.Context, cfg Config, newTeleport NewProcess) error {
	if newTeleport == nil {
		newTeleport = newTeleportProcess
	}
	copyCfg := cfg
	srv, err := newTeleport(&copyCfg)
	if err != nil {
		return trace.Wrap(err, "initialization failed")
	}
	if srv == nil {
		return trace.BadParameter("process has returned nil server")
	}
	if err := srv.Start(); err != nil {
		return trace.Wrap(err, "startup failed")
	}
	// Wait and reload until called exit.
	for {
		srv, err = waitAndReload(ctx, cfg, srv, newTeleport)
		if err != nil {
			// This error means that was a clean shutdown
			// and no reload is necessary.
			if err == ErrTeleportExited {
				return nil
			}
			return trace.Wrap(err)
		}
	}
}

func waitAndReload(ctx context.Context, cfg Config, srv Process, newTeleport NewProcess) (Process, error) {
	err := srv.WaitForSignals(ctx)
	if err == nil {
		return nil, ErrTeleportExited
	}
	if err != ErrTeleportReloading {
		return nil, trace.Wrap(err)
	}
	cfg.Log.Infof("Started in-process service reload.")
	fileDescriptors, err := srv.ExportFileDescriptors()
	if err != nil {
		warnOnErr(srv.Close(), cfg.Log)
		return nil, trace.Wrap(err)
	}
	newCfg := cfg
	newCfg.FileDescriptors = fileDescriptors
	newSrv, err := newTeleport(&newCfg)
	if err != nil {
		warnOnErr(srv.Close(), cfg.Log)
		return nil, trace.Wrap(err, "failed to create a new service")
	}
	cfg.Log.Infof("Created new process.")
	if err := newSrv.Start(); err != nil {
		warnOnErr(srv.Close(), cfg.Log)
		return nil, trace.Wrap(err, "failed to start a new service")
	}
	// Wait for the new server to report that it has started
	// before shutting down the old one.
	startTimeoutCtx, startCancel := context.WithTimeout(ctx, signalPipeTimeout)
	defer startCancel()
	eventC := make(chan Event, 1)
	newSrv.WaitForEvent(startTimeoutCtx, TeleportReadyEvent, eventC)
	select {
	case <-eventC:
		cfg.Log.Infof("New service has started successfully.")
	case <-startTimeoutCtx.Done():
		warnOnErr(newSrv.Close(), cfg.Log)
		warnOnErr(srv.Close(), cfg.Log)
		return nil, trace.BadParameter("the new service has failed to start")
	}
	shutdownTimeout := cfg.ShutdownTimeout
	if shutdownTimeout == 0 {
		// The default shutdown timeout is very generous to avoid disrupting
		// longer running connections.
		shutdownTimeout = defaults.DefaultGracefulShutdownTimeout
	}
	cfg.Log.Infof("Shutting down the old service with timeout %v.", shutdownTimeout)
	// After the new process has started, initiate the graceful shutdown of the old process
	// new process could have generated connections to the new process's server
	// so not all connections can be kept forever.
	timeoutCtx, cancel := context.WithTimeout(ctx, shutdownTimeout)
	defer cancel()
	srv.Shutdown(timeoutCtx)
	if timeoutCtx.Err() == context.DeadlineExceeded {
		// The new serivce can start initiating connections to the old service
		// keeping it from shutting down gracefully, or some external
		// connections can keep hanging the old auth service and prevent
		// the services from shutting down, so abort the graceful way
		// after some time to keep going.
		cfg.Log.Infof("Some connections to the old service were aborted after timeout of %v.", shutdownTimeout)
		// Make sure that all parts of the service have exited, this function
		// can not allow execution to continue if the shutdown is not complete,
		// otherwise subsequent Run executions will hold system resources in case
		// if old versions of the service are not exiting completely.
		timeoutCtx, cancel := context.WithTimeout(ctx, shutdownTimeout)
		defer cancel()
		srv.WaitWithContext(timeoutCtx)
		if timeoutCtx.Err() == context.DeadlineExceeded {
			return nil, trace.BadParameter("the old service has failed to exit.")
		}
	} else {
		cfg.Log.Infof("The old service was successfully shut down gracefully.")
	}
	return newSrv, nil
}

// NewTeleport takes the daemon configuration, instantiates all required services
// and starts them under a supervisor, returning the supervisor object.
func NewTeleport(cfg *Config) (*TeleportProcess, error) {
	var err error

	// Before we do anything reset the SIGINT handler back to the default.
	system.ResetInterruptSignalHandler()

	// Validate the config before accessing it.
	if err := validateConfig(cfg); err != nil {
		return nil, trace.Wrap(err, "configuration error")
	}

	// If FIPS mode was requested make sure binary is build against BoringCrypto.
	if cfg.FIPS {
		if !modules.GetModules().IsBoringBinary() {
			return nil, trace.BadParameter("binary not compiled against BoringCrypto, check " +
				"that Enterprise release was downloaded from " +
				"https://dashboard.gravitational.com")
		}
	}

	// create the data directory if it's missing
	_, err = os.Stat(cfg.DataDir)
	if os.IsNotExist(err) {
		err := os.MkdirAll(cfg.DataDir, os.ModeDir|0700)
		if err != nil {
			return nil, trace.ConvertSystemError(err)
		}
	}

	if len(cfg.FileDescriptors) == 0 {
		cfg.FileDescriptors, err = importFileDescriptors(cfg.Log)
		if err != nil {
			return nil, trace.Wrap(err)
		}
	}

	// if there's no host uuid initialized yet, try to read one from the
	// one of the identities
	cfg.HostUUID, err = utils.ReadHostUUID(cfg.DataDir)
	if err != nil {
		if !trace.IsNotFound(err) {
			return nil, trace.Wrap(err)
		}
		if len(cfg.Identities) != 0 {
			cfg.HostUUID = cfg.Identities[0].ID.HostUUID
			cfg.Log.Infof("Taking host UUID from first identity: %v.", cfg.HostUUID)
		} else {
			cfg.HostUUID = uuid.New()
			cfg.Log.Infof("Generating new host UUID: %v.", cfg.HostUUID)
		}
		if err := utils.WriteHostUUID(cfg.DataDir, cfg.HostUUID); err != nil {
			return nil, trace.Wrap(err)
		}
	}

	if uuid.Parse(cfg.HostUUID) == nil {
		cfg.Log.Warnf("Host UUID %q is not a true UUID (not eligible for UUID-based proxying)", cfg.HostUUID)
	}

	// if user started auth and another service (without providing the auth address for
	// that service, the address of the in-process auth will be used
	if cfg.Auth.Enabled && len(cfg.AuthServers) == 0 {
		cfg.AuthServers = []utils.NetAddr{cfg.Auth.SSHAddr}
	}

	// if user did not provide auth domain name, use this host's name
	if cfg.Auth.Enabled && cfg.Auth.ClusterName == nil {
		cfg.Auth.ClusterName, err = services.NewClusterNameWithRandomID(types.ClusterNameSpecV2{
			ClusterName: cfg.Hostname,
		})
		if err != nil {
			return nil, trace.Wrap(err)
		}
	}

	processID := fmt.Sprintf("%v", nextProcessID())
	supervisor := NewSupervisor(processID, cfg.Log)
	storage, err := auth.NewProcessStorage(supervisor.ExitContext(), filepath.Join(cfg.DataDir, teleport.ComponentProcess))
	if err != nil {
		return nil, trace.Wrap(err)
	}

	if cfg.Clock == nil {
		cfg.Clock = clockwork.NewRealClock()
	}

	if cfg.PluginRegistry == nil {
		cfg.PluginRegistry = plugin.NewRegistry()
	}

	process := &TeleportProcess{
		PluginRegistry:      cfg.PluginRegistry,
		Clock:               cfg.Clock,
		Supervisor:          supervisor,
		Config:              cfg,
		Identities:          make(map[types.SystemRole]*auth.Identity),
		connectors:          make(map[types.SystemRole]*Connector),
		importedDescriptors: cfg.FileDescriptors,
		storage:             storage,
		id:                  processID,
		keyPairs:            make(map[keyPairKey]KeyPair),
		appDependCh:         make(chan Event, 1024),
	}

	process.registerAppDepend()

	process.log = cfg.Log.WithFields(logrus.Fields{
		trace.Component: teleport.Component(teleport.ComponentProcess, process.id),
	})

	serviceStarted := false

	if !cfg.DiagnosticAddr.IsEmpty() {
		if err := process.initDiagnosticService(); err != nil {
			return nil, trace.Wrap(err)
		}
	} else {
		warnOnErr(process.closeImportedDescriptors(teleport.ComponentDiagnostic), process.log)
	}

	// Create a process wide key generator that will be shared. This is so the
	// key generator can pre-generate keys and share these across services.
	if cfg.Keygen == nil {
		precomputeCount := native.PrecomputedNum
		// in case if not auth or proxy services are enabled,
		// there is no need to precompute any SSH keys in the pool
		if !cfg.Auth.Enabled && !cfg.Proxy.Enabled {
			precomputeCount = 0
		}
		cfg.Keygen = native.New(process.ExitContext(), native.PrecomputeKeys(precomputeCount))
	}

	// Produce global TeleportReadyEvent
	// when all components have started
	eventMapping := EventMapping{
		Out: TeleportReadyEvent,
	}
	if cfg.Auth.Enabled {
		eventMapping.In = append(eventMapping.In, AuthTLSReady)
	}
	if cfg.SSH.Enabled {
		eventMapping.In = append(eventMapping.In, NodeSSHReady)
	}
	if cfg.Proxy.Enabled {
		eventMapping.In = append(eventMapping.In, ProxySSHReady)
	}
	if cfg.Kube.Enabled {
		eventMapping.In = append(eventMapping.In, KubernetesReady)
	}
	if cfg.Apps.Enabled {
		eventMapping.In = append(eventMapping.In, AppsReady)
	}
	process.RegisterEventMapping(eventMapping)

	if cfg.Auth.Enabled {
		if err := process.initAuthService(); err != nil {
			return nil, trace.Wrap(err)
		}
		serviceStarted = true
	} else {
		warnOnErr(process.closeImportedDescriptors(teleport.ComponentAuth), process.log)
	}

	if cfg.SSH.Enabled {
		if err := process.initSSH(); err != nil {
			return nil, err
		}
		serviceStarted = true
	} else {
		warnOnErr(process.closeImportedDescriptors(teleport.ComponentNode), process.log)
	}

	if cfg.Proxy.Enabled {
		if err := process.initProxy(); err != nil {
			return nil, err
		}
		serviceStarted = true
	} else {
		warnOnErr(process.closeImportedDescriptors(teleport.ComponentProxy), process.log)
	}

	if cfg.Kube.Enabled {
		process.initKubernetes()
		serviceStarted = true
	} else {
		warnOnErr(process.closeImportedDescriptors(teleport.ComponentKube), process.log)
	}

	// If this process is proxying applications, start application access server.
	if cfg.Apps.Enabled {
		process.initApps()
		serviceStarted = true
	} else {
		warnOnErr(process.closeImportedDescriptors(teleport.ComponentApp), process.log)
	}

	if cfg.Databases.Enabled {
		process.initDatabases()
		serviceStarted = true
	} else {
		warnOnErr(process.closeImportedDescriptors(teleport.ComponentDatabase), process.log)
	}

	process.RegisterFunc("common.rotate", process.periodicSyncRotationState)

	if !serviceStarted {
		return nil, trace.BadParameter("all services failed to start")
	}

	// create the new pid file only after started successfully
	if cfg.PIDFile != "" {
		f, err := os.OpenFile(cfg.PIDFile, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0666)
		if err != nil {
			return nil, trace.ConvertSystemError(err)
		}
		_, err = fmt.Fprintf(f, "%v", os.Getpid())
		if err = trace.NewAggregate(err, f.Close()); err != nil {
			return nil, trace.Wrap(err)
		}
	}

	// notify parent process that this process has started
	go process.notifyParent()

	return process, nil
}

// notifyParent notifies parent process that this process has started
// by writing to in-memory pipe used by communication channel.
func (process *TeleportProcess) notifyParent() {
	signalPipe, err := process.importSignalPipe()
	if err != nil {
		if !trace.IsNotFound(err) {
			process.log.Warningf("Failed to import signal pipe")
		}
		process.log.Debugf("No signal pipe to import, must be first Teleport process.")
		return
	}
	defer signalPipe.Close()

	ctx, cancel := context.WithTimeout(process.ExitContext(), signalPipeTimeout)
	defer cancel()

	eventC := make(chan Event, 1)
	process.WaitForEvent(ctx, TeleportReadyEvent, eventC)
	select {
	case <-eventC:
		process.log.Infof("New service has started successfully.")
	case <-ctx.Done():
		process.log.Errorf("Timeout waiting for a forked process to start: %v. Initiating self-shutdown.", ctx.Err())
		if err := process.Close(); err != nil {
			process.log.Warningf("Failed to shutdown process: %v.", err)
		}
		return
	}

	if err := process.writeToSignalPipe(signalPipe, fmt.Sprintf("Process %v has started.", os.Getpid())); err != nil {
		process.log.Warningf("Failed to write to signal pipe: %v", err)
		// despite the failure, it's ok to proceed,
		// it could mean that the parent process has crashed and the pipe
		// is no longer valid.
	}
}

func (process *TeleportProcess) setLocalAuth(a *auth.Server) {
	process.Lock()
	defer process.Unlock()
	process.localAuth = a
}

func (process *TeleportProcess) getLocalAuth() *auth.Server {
	process.Lock()
	defer process.Unlock()
	return process.localAuth
}

// adminCreds returns admin UID and GID settings based on the OS
func adminCreds() (*int, *int, error) {
	if runtime.GOOS != constants.LinuxOS {
		return nil, nil, nil
	}
	// if the user member of adm linux group,
	// make audit log folder readable by admins
	isAdmin, err := utils.IsGroupMember(teleport.LinuxAdminGID)
	if err != nil {
		return nil, nil, trace.Wrap(err)
	}
	if !isAdmin {
		return nil, nil, nil
	}
	uid := os.Getuid()
	gid := teleport.LinuxAdminGID
	return &uid, &gid, nil
}

// initUploadHandler initializes upload handler based on the config settings,
// currently the only upload handler supported is S3
// the call can return trace.NotFound if no upload handler is setup
func initUploadHandler(auditConfig types.ClusterAuditConfig, dataDir string) (events.MultipartHandler, error) {
	if !auditConfig.ShouldUploadSessions() {
		recordsDir := filepath.Join(dataDir, events.RecordsDir)
		if err := os.MkdirAll(recordsDir, teleport.SharedDirMode); err != nil {
			return nil, trace.ConvertSystemError(err)
		}
		handler, err := filesessions.NewHandler(filesessions.Config{
			Directory: recordsDir,
		})
		if err != nil {
			return nil, trace.Wrap(err)
		}
		wrapper, err := events.NewLegacyHandler(events.LegacyHandlerConfig{
			Handler: handler,
			Dir:     dataDir,
		})
		if err != nil {
			return nil, trace.Wrap(err)
		}
		return wrapper, nil
	}
	uri, err := utils.ParseSessionsURI(auditConfig.AuditSessionsURI())
	if err != nil {
		return nil, trace.Wrap(err)
	}

	switch uri.Scheme {
	case teleport.SchemeGCS:
		config := gcssessions.Config{}
		if err := config.SetFromURL(uri); err != nil {
			return nil, trace.Wrap(err)
		}
		handler, err := gcssessions.DefaultNewHandler(config)
		if err != nil {
			return nil, trace.Wrap(err)
		}
		return handler, nil
	case teleport.SchemeS3:
		config := s3sessions.Config{}
		if err := config.SetFromURL(uri, auditConfig.Region()); err != nil {
			return nil, trace.Wrap(err)
		}
		handler, err := s3sessions.NewHandler(config)
		if err != nil {
			return nil, trace.Wrap(err)
		}
		return handler, nil
	case teleport.SchemeFile:
		if err := os.MkdirAll(uri.Path, teleport.SharedDirMode); err != nil {
			return nil, trace.ConvertSystemError(err)
		}
		handler, err := filesessions.NewHandler(filesessions.Config{
			Directory: uri.Path,
		})
		if err != nil {
			return nil, trace.Wrap(err)
		}
		return handler, nil
	default:
		return nil, trace.BadParameter(
			"unsupported scheme for audit_sesions_uri: %q, currently supported schemes are %q and %q",
			uri.Scheme, teleport.SchemeS3, teleport.SchemeFile)
	}
}

// initExternalLog initializes external storage, if the storage is not
// setup, returns (nil, nil).
func initExternalLog(ctx context.Context, auditConfig types.ClusterAuditConfig, log logrus.FieldLogger, backend backend.Backend) (events.IAuditLog, error) {
	var hasNonFileLog bool
	var loggers []events.IAuditLog
	for _, eventsURI := range auditConfig.AuditEventsURIs() {
		uri, err := utils.ParseSessionsURI(eventsURI)
		if err != nil {
			return nil, trace.Wrap(err)
		}
		switch uri.Scheme {
		case firestore.GetName():
			hasNonFileLog = true
			cfg := firestoreevents.EventsConfig{}
			err = cfg.SetFromURL(uri)
			if err != nil {
				return nil, trace.Wrap(err)
			}
			logger, err := firestoreevents.New(cfg)
			if err != nil {
				return nil, trace.Wrap(err)
			}
			loggers = append(loggers, logger)
		case dynamo.GetName():
			hasNonFileLog = true
			cfg := dynamoevents.Config{
				Tablename:               uri.Host,
				Region:                  auditConfig.Region(),
				EnableContinuousBackups: auditConfig.EnableContinuousBackups(),
				EnableAutoScaling:       auditConfig.EnableAutoScaling(),
				ReadMinCapacity:         auditConfig.ReadMinCapacity(),
				ReadMaxCapacity:         auditConfig.ReadMaxCapacity(),
				ReadTargetValue:         auditConfig.ReadTargetValue(),
				WriteMinCapacity:        auditConfig.WriteMinCapacity(),
				WriteMaxCapacity:        auditConfig.WriteMaxCapacity(),
				WriteTargetValue:        auditConfig.WriteTargetValue(),
			}
			err = cfg.SetFromURL(uri)
			if err != nil {
				return nil, trace.Wrap(err)
			}

			logger, err := dynamoevents.New(ctx, cfg, backend)
			if err != nil {
				return nil, trace.Wrap(err)
			}
			loggers = append(loggers, logger)
		case teleport.SchemeFile:
			if uri.Path == "" {
				return nil, trace.BadParameter("unsupported audit uri: %q (missing path component)", uri)
			}
			if uri.Host != "" && uri.Host != "localhost" {
				return nil, trace.BadParameter("unsupported audit uri: %q (nonlocal host component: %q)", uri, uri.Host)
			}
			if err := os.MkdirAll(uri.Path, teleport.SharedDirMode); err != nil {
				return nil, trace.ConvertSystemError(err)
			}
			logger, err := events.NewFileLog(events.FileLogConfig{
				Dir: uri.Path,
			})
			if err != nil {
				return nil, trace.Wrap(err)
			}
			loggers = append(loggers, logger)
		case teleport.SchemeStdout:
			logger := events.NewWriterEmitter(utils.NopWriteCloser(os.Stdout))
			loggers = append(loggers, logger)
		default:
			return nil, trace.BadParameter(
				"unsupported scheme for audit_events_uri: %q, currently supported schemes are %q and %q",
				uri.Scheme, dynamo.GetName(), teleport.SchemeFile)
		}
	}

	if len(loggers) < 1 {
		return nil, nil
	}

	if !auditConfig.ShouldUploadSessions() && hasNonFileLog {
		// if audit events are being exported, session recordings should
		// be exported as well.
		return nil, trace.BadParameter("please specify audit_sessions_uri when using external audit backends")
	}

	if len(loggers) > 1 {
		return events.NewMultiLog(loggers...)
	}

	return loggers[0], nil
}

// initAuthService can be called to initialize auth server service
func (process *TeleportProcess) initAuthService() error {
	var err error

	cfg := process.Config

	// Initialize the storage back-ends for keys, events and records
	b, err := process.initAuthStorage()
	if err != nil {
		return trace.Wrap(err)
	}
	process.backend = b

	var emitter apievents.Emitter
	var streamer events.Streamer
	var uploadHandler events.MultipartHandler
	// create the audit log, which will be consuming (and recording) all events
	// and recording all sessions.
	if cfg.Auth.NoAudit {
		// this is for teleconsole
		process.auditLog = events.NewDiscardAuditLog()

		warningMessage := "Warning: Teleport audit and session recording have been " +
			"turned off. This is dangerous, you will not be able to view audit events " +
			"or save and playback recorded sessions."
		process.log.Warn(warningMessage)
		discard := events.NewDiscardEmitter()
		emitter, streamer = discard, discard
	} else {
		// check if session recording has been disabled. note, we will continue
		// logging audit events, we just won't record sessions.
		recordSessions := true
		if cfg.Auth.SessionRecordingConfig.GetMode() == types.RecordOff {
			recordSessions = false

			warningMessage := "Warning: Teleport session recording have been turned off. " +
				"This is dangerous, you will not be able to save and playback sessions."
			process.log.Warn(warningMessage)
		}

		auditConfig := cfg.Auth.AuditConfig
		uploadHandler, err = initUploadHandler(
			auditConfig, filepath.Join(cfg.DataDir, teleport.LogsDir))
		if err != nil {
			if !trace.IsNotFound(err) {
				return trace.Wrap(err)
			}
		}
		streamer, err = events.NewProtoStreamer(events.ProtoStreamerConfig{
			Uploader: uploadHandler,
		})
		if err != nil {
			return trace.Wrap(err)
		}
		// initialize external loggers.  may return (nil, nil) if no
		// external loggers have been defined.
		externalLog, err := initExternalLog(process.ExitContext(), auditConfig, process.log, process.backend)
		if err != nil {
			if !trace.IsNotFound(err) {
				return trace.Wrap(err)
			}
		}

		auditServiceConfig := events.AuditLogConfig{
			Context:        process.ExitContext(),
			DataDir:        filepath.Join(cfg.DataDir, teleport.LogsDir),
			RecordSessions: recordSessions,
			ServerID:       cfg.HostUUID,
			UploadHandler:  uploadHandler,
			ExternalLog:    externalLog,
		}
		auditServiceConfig.UID, auditServiceConfig.GID, err = adminCreds()
		if err != nil {
			return trace.Wrap(err)
		}
		localLog, err := events.NewAuditLog(auditServiceConfig)
		if err != nil {
			return trace.Wrap(err)
		}
		process.auditLog = localLog
		if externalLog != nil {
			externalEmitter, ok := externalLog.(apievents.Emitter)
			if !ok {
				return trace.BadParameter("expected emitter, but %T does not emit", externalLog)
			}
			emitter = externalEmitter
		} else {
			emitter = localLog
		}
	}

	// Upload completer is responsible for checking for initiated but abandoned
	// session uploads and completing them
	var uploadCompleter *events.UploadCompleter
	if uploadHandler != nil {
		uploadCompleter, err = events.NewUploadCompleter(events.UploadCompleterConfig{
			Uploader:  uploadHandler,
			Component: teleport.ComponentAuth,
			AuditLog:  process.auditLog,
		})
		if err != nil {
			return trace.Wrap(err)
		}
	}

	checkingEmitter, err := events.NewCheckingEmitter(events.CheckingEmitterConfig{
		Inner:       events.NewMultiEmitter(events.NewLoggingEmitter(), emitter),
		Clock:       process.Clock,
		ClusterName: cfg.Auth.ClusterName.GetClusterName(),
	})
	if err != nil {
		return trace.Wrap(err)
	}

	checkingStreamer, err := events.NewCheckingStreamer(events.CheckingStreamerConfig{
		Inner:       streamer,
		Clock:       process.Clock,
		ClusterName: cfg.Auth.ClusterName.GetClusterName(),
	})
	if err != nil {
		return trace.Wrap(err)
	}

	// first, create the AuthServer
	authServer, err := auth.Init(auth.InitConfig{
		Backend:                 b,
		Authority:               cfg.Keygen,
		ClusterConfiguration:    cfg.ClusterConfiguration,
		ClusterConfig:           cfg.Auth.ClusterConfig,
		ClusterAuditConfig:      cfg.Auth.AuditConfig,
		ClusterNetworkingConfig: cfg.Auth.NetworkingConfig,
		SessionRecordingConfig:  cfg.Auth.SessionRecordingConfig,
		ClusterName:             cfg.Auth.ClusterName,
		AuthServiceName:         cfg.Hostname,
		DataDir:                 cfg.DataDir,
		HostUUID:                cfg.HostUUID,
		NodeName:                cfg.Hostname,
		Authorities:             cfg.Auth.Authorities,
		Resources:               cfg.Auth.Resources,
		ReverseTunnels:          cfg.ReverseTunnels,
		Trust:                   cfg.Trust,
		Presence:                cfg.Presence,
		Events:                  cfg.Events,
		Provisioner:             cfg.Provisioner,
		Identity:                cfg.Identity,
		Access:                  cfg.Access,
		StaticTokens:            cfg.Auth.StaticTokens,
		Roles:                   cfg.Auth.Roles,
		AuthPreference:          cfg.Auth.Preference,
		OIDCConnectors:          cfg.OIDCConnectors,
		AuditLog:                process.auditLog,
		CipherSuites:            cfg.CipherSuites,
		CASigningAlg:            cfg.CASignatureAlgorithm,
		Emitter:                 checkingEmitter,
		Streamer:                events.NewReportingStreamer(checkingStreamer, process.Config.UploadEventsC),
	})
	if err != nil {
		return trace.Wrap(err)
	}

	process.setLocalAuth(authServer)

	connector, err := process.connectToAuthService(types.RoleAdmin)
	if err != nil {
		return trace.Wrap(err)
	}

	// second, create the API Server: it's actually a collection of API servers,
	// each serving requests for a "role" which is assigned to every connected
	// client based on their certificate (user, server, admin, etc)
	sessionService, err := session.New(b)
	if err != nil {
		return trace.Wrap(err)
	}
	authorizer, err := auth.NewAuthorizer(cfg.Auth.ClusterName.GetClusterName(), authServer.Access, authServer.Identity, authServer.Trust)
	if err != nil {
		return trace.Wrap(err)
	}
	apiConf := &auth.APIConfig{
		AuthServer:     authServer,
		SessionService: sessionService,
		Authorizer:     authorizer,
		AuditLog:       process.auditLog,
		PluginRegistry: process.PluginRegistry,
		Emitter:        checkingEmitter,
		MetadataGetter: uploadHandler,
	}

	var authCache auth.Cache
	if process.Config.CachePolicy.Enabled {
		cache, err := process.newAccessCache(accessCacheConfig{
			services:  authServer.Services,
			setup:     cache.ForAuth,
			cacheName: []string{teleport.ComponentAuth},
			inMemory:  true,
			events:    true,
		})
		if err != nil {
			return trace.Wrap(err)
		}
		authCache = cache
	} else {
		authCache = authServer.Services
	}
	authServer.SetCache(authCache)

	log := process.log.WithFields(logrus.Fields{
		trace.Component: teleport.Component(teleport.ComponentAuth, process.id),
	})

	// Register TLS endpoint of the auth service
	tlsConfig, err := connector.ServerIdentity.TLSConfig(cfg.CipherSuites)
	if err != nil {
		return trace.Wrap(err)
	}
	// auth server listens on SSH and TLS, reusing the same socket
	listener, err := process.importOrCreateListener(listenerAuthSSH, cfg.Auth.SSHAddr.Addr)
	if err != nil {
		log.Errorf("PID: %v Failed to bind to address %v: %v, exiting.", os.Getpid(), cfg.Auth.SSHAddr.Addr, err)
		return trace.Wrap(err)
	}

	// use listener addr instead of cfg.Auth.SSHAddr in order to support
	// binding to a random port (e.g. `127.0.0.1:0`).
	authAddr := listener.Addr().String()

	// clean up unused descriptors passed for proxy, but not used by it
	warnOnErr(process.closeImportedDescriptors(teleport.ComponentAuth), log)
	if cfg.Auth.EnableProxyProtocol {
		log.Infof("Starting Auth service with PROXY protocol support.")
	}
	mux, err := multiplexer.New(multiplexer.Config{
		EnableProxyProtocol: cfg.Auth.EnableProxyProtocol,
		Listener:            listener,
		ID:                  teleport.Component(process.id),
	})
	if err != nil {
		listener.Close()
		return trace.Wrap(err)
	}
	go mux.Serve()
	tlsServer, err := auth.NewTLSServer(auth.TLSServerConfig{
		TLS:           tlsConfig,
		APIConfig:     *apiConf,
		LimiterConfig: cfg.Auth.Limiter,
		AccessPoint:   authCache,
		Component:     teleport.Component(teleport.ComponentAuth, process.id),
		ID:            process.id,
		Listener:      mux.TLS(),
	})
	if err != nil {
		return trace.Wrap(err)
	}
	process.RegisterCriticalFunc("auth.tls", func() error {
		utils.Consolef(cfg.Console, log, teleport.ComponentAuth, "Auth service %s:%s is starting on %v.",
			teleport.Version, teleport.Gitref, authAddr)

		// since tlsServer.Serve is a blocking call, we emit this even right before
		// the service has started
		process.BroadcastEvent(Event{Name: AuthTLSReady, Payload: nil})
		err := tlsServer.Serve()
		if err != nil && err != http.ErrServerClosed {
			log.Warningf("TLS server exited with error: %v.", err)
		}
		return nil
	})
	process.RegisterFunc("auth.heartbeat.broadcast", func() error {
		// Heart beat auth server presence, this is not the best place for this
		// logic, consolidate it into auth package later
		connector, err := process.connectToAuthService(types.RoleAdmin)
		if err != nil {
			return trace.Wrap(err)
		}
		// External integrations rely on this event:
		process.BroadcastEvent(Event{Name: AuthIdentityEvent, Payload: connector})
		process.OnExit("auth.broadcast", func(payload interface{}) {
			connector.Close()
		})
		return nil
	})

	host, port, err := net.SplitHostPort(authAddr)
	if err != nil {
		return trace.Wrap(err)
	}
	// advertise-ip is explicitly set:
	if process.Config.AdvertiseIP != "" {
		ahost, aport, err := utils.ParseAdvertiseAddr(process.Config.AdvertiseIP)
		if err != nil {
			return trace.Wrap(err)
		}
		// if port is not set in the advertise addr, use the default one
		if aport == "" {
			aport = port
		}
		authAddr = net.JoinHostPort(ahost, aport)
	} else {
		// advertise-ip is not set, while the CA is listening on 0.0.0.0? lets try
		// to guess the 'advertise ip' then:
		if net.ParseIP(host).IsUnspecified() {
			ip, err := utils.GuessHostIP()
			if err != nil {
				log.Warn(err)
			} else {
				authAddr = net.JoinHostPort(ip.String(), port)
			}
		}
		log.Warnf("Configuration setting auth_service/advertise_ip is not set. guessing %v.", authAddr)
	}

	heartbeat, err := srv.NewHeartbeat(srv.HeartbeatConfig{
		Mode:      srv.HeartbeatModeAuth,
		Context:   process.ExitContext(),
		Component: teleport.ComponentAuth,
		Announcer: authServer,
		GetServerInfo: func() (types.Resource, error) {
			srv := types.ServerV2{
				Kind:    types.KindAuthServer,
				Version: types.V2,
				Metadata: types.Metadata{
					Namespace: apidefaults.Namespace,
					Name:      process.Config.HostUUID,
				},
				Spec: types.ServerSpecV2{
					Addr:     authAddr,
					Hostname: process.Config.Hostname,
					Version:  teleport.Version,
				},
			}
			state, err := process.storage.GetState(types.RoleAdmin)
			if err != nil {
				if !trace.IsNotFound(err) {
					log.Warningf("Failed to get rotation state: %v.", err)
					return nil, trace.Wrap(err)
				}
			} else {
				srv.Spec.Rotation = state.Spec.Rotation
			}
			srv.SetExpiry(process.Clock.Now().UTC().Add(apidefaults.ServerAnnounceTTL))
			return &srv, nil
		},
		KeepAlivePeriod: apidefaults.ServerKeepAliveTTL,
		AnnouncePeriod:  apidefaults.ServerAnnounceTTL/2 + utils.RandomDuration(apidefaults.ServerAnnounceTTL/10),
		CheckPeriod:     defaults.HeartbeatCheckPeriod,
		ServerTTL:       apidefaults.ServerAnnounceTTL,
		OnHeartbeat: func(err error) {
			if err != nil {
				process.BroadcastEvent(Event{Name: TeleportDegradedEvent, Payload: teleport.ComponentAuth})
			} else {
				process.BroadcastEvent(Event{Name: TeleportOKEvent, Payload: teleport.ComponentAuth})
			}
		},
	})
	if err != nil {
		return trace.Wrap(err)
	}
	process.RegisterFunc("auth.heartbeat", heartbeat.Run)
	// execute this when process is asked to exit:
	process.OnExit("auth.shutdown", func(payload interface{}) {
		// The listeners have to be closed here, because if shutdown
		// was called before the start of the http server,
		// the http server would have not started tracking the listeners
		// and http.Shutdown will do nothing.
		if mux != nil {
			warnOnErr(mux.Close(), log)
		}
		if listener != nil {
			warnOnErr(listener.Close(), log)
		}
		if payload == nil {
			log.Info("Shutting down immediately.")
			warnOnErr(tlsServer.Close(), log)
		} else {
			log.Info("Shutting down gracefully.")
			ctx := payloadContext(payload, log)
			warnOnErr(tlsServer.Shutdown(ctx), log)
		}
		if uploadCompleter != nil {
			warnOnErr(uploadCompleter.Close(), log)
		}
		log.Info("Exited.")
	})
	return nil
}

func payloadContext(payload interface{}, log logrus.FieldLogger) context.Context {
	ctx, ok := payload.(context.Context)
	if ok {
		return ctx
	}
	log.Errorf("Expected context, got %T.", payload)
	return context.TODO()
}

// OnExit allows individual services to register a callback function which will be
// called when Teleport Process is asked to exit. Usually services terminate themselves
// when the callback is called
func (process *TeleportProcess) OnExit(serviceName string, callback func(interface{})) {
	process.RegisterFunc(serviceName, func() error {
		eventC := make(chan Event)
		process.WaitForEvent(context.TODO(), TeleportExitEvent, eventC)

		event := <-eventC
		callback(event.Payload)
		return nil
	})
}

// accessCacheConfig contains
// configuration for access cache
type accessCacheConfig struct {
	// services is a collection
	// of services to use as a cache base
	services services.Services
	// setup is a function that takes
	// cache configuration and modifies it
	setup cache.SetupConfigFn
	// cacheName is a cache name
	cacheName []string
	// inMemory is true if cache
	// should use memory
	inMemory bool
	// events is true if cache should turn on events
	events bool
	// pollPeriod contains period for polling
	pollPeriod time.Duration
}

func (c *accessCacheConfig) CheckAndSetDefaults() error {
	if c.services == nil {
		return trace.BadParameter("missing parameter services")
	}
	if c.setup == nil {
		return trace.BadParameter("missing parameter setup")
	}
	if len(c.cacheName) == 0 {
		return trace.BadParameter("missing parameter cacheName")
	}
	if c.pollPeriod == 0 {
		c.pollPeriod = defaults.CachePollPeriod
	}
	return nil
}

// newAccessCache returns new local cache access point
func (process *TeleportProcess) newAccessCache(cfg accessCacheConfig) (*cache.Cache, error) {
	if err := cfg.CheckAndSetDefaults(); err != nil {
		return nil, trace.Wrap(err)
	}
	var cacheBackend backend.Backend
	if cfg.inMemory {
		process.log.Debugf("Creating in-memory backend for %v.", cfg.cacheName)
		mem, err := memory.New(memory.Config{
			Context:   process.ExitContext(),
			EventsOff: !cfg.events,
			Mirror:    true,
		})
		if err != nil {
			return nil, trace.Wrap(err)
		}
		cacheBackend = mem
	} else {
		process.log.Debugf("Creating sqlite backend for %v.", cfg.cacheName)
		path := filepath.Join(append([]string{process.Config.DataDir, "cache"}, cfg.cacheName...)...)
		if err := os.MkdirAll(path, teleport.SharedDirMode); err != nil {
			return nil, trace.ConvertSystemError(err)
		}
		liteBackend, err := lite.NewWithConfig(process.ExitContext(),
			lite.Config{
				Path:             path,
				EventsOff:        !cfg.events,
				Memory:           false,
				Mirror:           true,
				PollStreamPeriod: 100 * time.Millisecond,
			})
		if err != nil {
			return nil, trace.Wrap(err)
		}
		cacheBackend = liteBackend
	}
	reporter, err := backend.NewReporter(backend.ReporterConfig{
		Component: teleport.ComponentCache,
		Backend:   cacheBackend,
	})
	if err != nil {
		return nil, trace.Wrap(err)
	}

	return cache.New(cfg.setup(cache.Config{
		Context:         process.ExitContext(),
		Backend:         reporter,
		Events:          cfg.services,
		ClusterConfig:   cfg.services,
		Provisioner:     cfg.services,
		Trust:           cfg.services,
		Users:           cfg.services,
		Access:          cfg.services,
		DynamicAccess:   cfg.services,
		Presence:        cfg.services,
		Restrictions:    cfg.services,
		AppSession:      cfg.services,
		WebSession:      cfg.services.WebSessions(),
		WebToken:        cfg.services.WebTokens(),
		Component:       teleport.Component(append(cfg.cacheName, process.id, teleport.ComponentCache)...),
		MetricComponent: teleport.Component(append(cfg.cacheName, teleport.ComponentCache)...),
	}))
}

// setupCachePolicy sets up cache policy based on teleport configuration,
// it is a wrapper function, that sets up configuration
func (process *TeleportProcess) setupCachePolicy(in cache.SetupConfigFn) cache.SetupConfigFn {
	return func(c cache.Config) cache.Config {
		config := in(c)
		config.PreferRecent = cache.PreferRecent{
			Enabled:      process.Config.CachePolicy.Enabled,
			NeverExpires: process.Config.CachePolicy.NeverExpires,
			MaxTTL:       process.Config.CachePolicy.TTL,
		}
		return config
	}
}

// newLocalCacheForProxy returns new instance of access point configured for a local proxy.
func (process *TeleportProcess) newLocalCacheForProxy(clt auth.ClientI, cacheName []string) (auth.AccessPoint, error) {
	return process.newLocalCache(clt, cache.ForProxy, cacheName)
}

// newLocalCacheForRemoteProxy returns new instance of access point configured
// for a remote proxy.
func (process *TeleportProcess) newLocalCacheForRemoteProxy(clt auth.ClientI, cacheName []string) (auth.AccessPoint, error) {
	return process.newLocalCache(clt, cache.ForRemoteProxy, cacheName)
}

// DELETE IN: 5.1.
//
// newLocalCacheForOldRemoteProxy returns new instance of access point
// configured for an old remote proxy.
func (process *TeleportProcess) newLocalCacheForOldRemoteProxy(clt auth.ClientI, cacheName []string) (auth.AccessPoint, error) {
	return process.newLocalCache(clt, cache.ForOldRemoteProxy, cacheName)
}

// newLocalCache returns new instance of access point
func (process *TeleportProcess) newLocalCache(clt auth.ClientI, setupConfig cache.SetupConfigFn, cacheName []string) (auth.AccessPoint, error) {
	// if caching is disabled, return access point
	if !process.Config.CachePolicy.Enabled {
		return clt, nil
	}
	cache, err := process.newAccessCache(accessCacheConfig{
		inMemory:  process.Config.CachePolicy.Type == memory.GetName(),
		services:  clt,
		setup:     process.setupCachePolicy(setupConfig),
		cacheName: cacheName,
	})
	if err != nil {
		return nil, trace.Wrap(err)
	}
	return auth.NewWrapper(clt, cache), nil
}

func (process *TeleportProcess) getRotation(role types.SystemRole) (*types.Rotation, error) {
	state, err := process.storage.GetState(role)
	if err != nil {
		return nil, trace.Wrap(err)
	}
	return &state.Spec.Rotation, nil
}

func (process *TeleportProcess) proxyPublicAddr() utils.NetAddr {
	if len(process.Config.Proxy.PublicAddrs) == 0 {
		return utils.NetAddr{}
	}
	return process.Config.Proxy.PublicAddrs[0]
}

// newAsyncEmitter wraps client and returns emitter that never blocks, logs some events and checks values.
// It is caller's responsibility to call Close on the emitter once done.
func (process *TeleportProcess) newAsyncEmitter(clt apievents.Emitter) (*events.AsyncEmitter, error) {
	emitter, err := events.NewCheckingEmitter(events.CheckingEmitterConfig{
		Inner: events.NewMultiEmitter(events.NewLoggingEmitter(), clt),
		Clock: process.Clock,
	})
	if err != nil {
		return nil, trace.Wrap(err)
	}

	// asyncEmitter makes sure that sessions do not block
	// in case if connections are slow
	return events.NewAsyncEmitter(events.AsyncEmitterConfig{
		Inner: emitter,
	})
}

// initSSH initializes the "node" role, i.e. a simple SSH server connected to the auth server.
func (process *TeleportProcess) initSSH() error {
	process.registerWithAuthServer(types.RoleNode, SSHIdentityEvent)
	eventsC := make(chan Event)
	process.WaitForEvent(process.ExitContext(), SSHIdentityEvent, eventsC)

	log := process.log.WithFields(logrus.Fields{
		trace.Component: teleport.Component(teleport.ComponentNode, process.id),
	})

	var agentPool *reversetunnel.AgentPool
	var conn *Connector
	var ebpf bpf.BPF
	var rm restricted.Manager
	var s *regular.Server
	var asyncEmitter *events.AsyncEmitter

	process.RegisterCriticalFunc("ssh.node", func() error {
		var ok bool
		var event Event

		select {
		case event = <-eventsC:
			log.Debugf("Received event %q.", event.Name)
		case <-process.ExitContext().Done():
			log.Debugf("Process is exiting.")
			return nil
		}

		conn, ok = (event.Payload).(*Connector)
		if !ok {
			return trace.BadParameter("unsupported connector type: %T", event.Payload)
		}

		cfg := process.Config

		limiter, err := limiter.NewLimiter(cfg.SSH.Limiter)
		if err != nil {
			return trace.Wrap(err)
		}

		authClient, err := process.newLocalCache(conn.Client, cache.ForNode, []string{teleport.ComponentNode})
		if err != nil {
			return trace.Wrap(err)
		}

		// If session recording is disabled at the cluster level and the node is
		// attempting to enabled enhanced session recording, show an error.
		recConfig, err := authClient.GetSessionRecordingConfig(process.ExitContext())
		if err != nil {
			return trace.Wrap(err)
		}
		if recConfig.GetMode() == types.RecordOff && cfg.SSH.BPF.Enabled {
			return trace.BadParameter("session recording is disabled at the cluster " +
				"level. To enable enhanced session recording, enable session recording at " +
				"the cluster level, then restart Teleport.")
		}

		// Restricted session requires BPF (enhanced recording)
		if cfg.SSH.RestrictedSession.Enabled && !cfg.SSH.BPF.Enabled {
			return trace.BadParameter("restricted_session requires enhanced_recording " +
				"to be enabled")
		}

		// If BPF is enabled in file configuration, but the operating system does
		// not support enhanced session recording (like macOS), exit right away.
		if cfg.SSH.BPF.Enabled && !bpf.SystemHasBPF() {
			return trace.BadParameter("operating system does not support enhanced " +
				"session recording, check Teleport documentation for more details on " +
				"supported operating systems, kernels, and configuration")
		}

		// Start BPF programs. This is blocking and if the BPF programs fail to
		// load, the node will not start. If BPF is not enabled, this will simply
		// return a NOP struct that can be used to discard BPF data.
		ebpf, err = bpf.New(cfg.SSH.BPF)
		if err != nil {
			return trace.Wrap(err)
		}

		// Start access control programs. This is blocking and if the BPF programs fail to
		// load, the node will not start. If access control is not enabled, this will simply
		// return a NOP struct.
		rm, err = restricted.New(cfg.SSH.RestrictedSession, conn.Client)
		if err != nil {
			return trace.Wrap(err)
		}

		// make sure the namespace exists
		namespace := types.ProcessNamespace(cfg.SSH.Namespace)
		_, err = authClient.GetNamespace(namespace)
		if err != nil {
			if trace.IsNotFound(err) {
				return trace.NotFound(
					"namespace %v is not found, ask your system administrator to create this namespace so you can register nodes there.", namespace)
			}
			return trace.Wrap(err)
		}

		// Provide helpful log message if listen_addr or public_addr are not being
		// used (tunnel is used to connect to cluster).
		//
		// If a tunnel is not being used, set the default here (could not be done in
		// file configuration because at that time it's not known if server is
		// joining cluster directly or through a tunnel).
		if conn.UseTunnel() {
			if !cfg.SSH.Addr.IsEmpty() {
				log.Info("Connected to cluster over tunnel connection, ignoring listen_addr setting.")
			}
			if len(cfg.SSH.PublicAddrs) > 0 {
				log.Info("Connected to cluster over tunnel connection, ignoring public_addr setting.")
			}
		}
		if !conn.UseTunnel() && cfg.SSH.Addr.IsEmpty() {
			cfg.SSH.Addr = *defaults.SSHServerListenAddr()
		}

		// asyncEmitter makes sure that sessions do not block
		// in case if connections are slow
		asyncEmitter, err = process.newAsyncEmitter(conn.Client)
		if err != nil {
			return trace.Wrap(err)
		}

		clusterName, err := authClient.GetClusterName()
		if err != nil {
			return trace.Wrap(err)
		}

		streamer, err := events.NewCheckingStreamer(events.CheckingStreamerConfig{
			Inner:       conn.Client,
			Clock:       process.Clock,
			ClusterName: clusterName.GetClusterName(),
		})
		if err != nil {
			return trace.Wrap(err)
		}

		s, err = regular.New(cfg.SSH.Addr,
			cfg.Hostname,
			[]ssh.Signer{conn.ServerIdentity.KeySigner},
			authClient,
			cfg.DataDir,
			cfg.AdvertiseIP,
			process.proxyPublicAddr(),
			regular.SetLimiter(limiter),
			regular.SetShell(cfg.SSH.Shell),
			regular.SetEmitter(&events.StreamerAndEmitter{Emitter: asyncEmitter, Streamer: streamer}),
			regular.SetSessionServer(conn.Client),
			regular.SetLabels(cfg.SSH.Labels, cfg.SSH.CmdLabels),
			regular.SetNamespace(namespace),
			regular.SetPermitUserEnvironment(cfg.SSH.PermitUserEnvironment),
			regular.SetCiphers(cfg.Ciphers),
			regular.SetKEXAlgorithms(cfg.KEXAlgorithms),
			regular.SetMACAlgorithms(cfg.MACAlgorithms),
			regular.SetPAMConfig(cfg.SSH.PAM),
			regular.SetRotationGetter(process.getRotation),
			regular.SetUseTunnel(conn.UseTunnel()),
			regular.SetFIPS(cfg.FIPS),
			regular.SetBPF(ebpf),
			regular.SetRestrictedSessionManager(rm),
			regular.SetOnHeartbeat(func(err error) {
				if err != nil {
					process.BroadcastEvent(Event{Name: TeleportDegradedEvent, Payload: teleport.ComponentNode})
				} else {
					process.BroadcastEvent(Event{Name: TeleportOKEvent, Payload: teleport.ComponentNode})
				}
			}),
			regular.SetAllowTCPForwarding(cfg.SSH.AllowTCPForwarding),
		)
		if err != nil {
			return trace.Wrap(err)
		}

		// init uploader service for recording SSH node, if proxy is not
		// enabled on this node, because proxy stars uploader service as well
		if !cfg.Proxy.Enabled {
			if err := process.initUploaderService(authClient, conn.Client); err != nil {
				return trace.Wrap(err)
			}
		}

		if !conn.UseTunnel() {
			listener, err := process.importOrCreateListener(listenerNodeSSH, cfg.SSH.Addr.Addr)
			if err != nil {
				return trace.Wrap(err)
			}
			// clean up unused descriptors passed for proxy, but not used by it
			warnOnErr(process.closeImportedDescriptors(teleport.ComponentNode), log)

			log.Infof("Service %s:%s is starting on %v %v.", teleport.Version, teleport.Gitref, cfg.SSH.Addr.Addr, process.Config.CachePolicy)
			utils.Consolef(cfg.Console, log, teleport.ComponentNode, "Service %s:%s is starting on %v.",
				teleport.Version, teleport.Gitref, cfg.SSH.Addr.Addr)

			// Start the SSH server. This kicks off updating labels, starting the
			// heartbeat, and accepting connections.
			go s.Serve(listener)

			// Broadcast that the node has started.
			process.BroadcastEvent(Event{Name: NodeSSHReady, Payload: nil})
		} else {
			// Start the SSH server. This kicks off updating labels and starting the
			// heartbeat.
			if err := s.Start(); err != nil {
				return trace.Wrap(err)
			}

			// Create and start an agent pool.
			agentPool, err = reversetunnel.NewAgentPool(
				process.ExitContext(),
				reversetunnel.AgentPoolConfig{
					Component:   teleport.ComponentNode,
					HostUUID:    conn.ServerIdentity.ID.HostUUID,
					ProxyAddr:   conn.TunnelProxy(),
					Client:      conn.Client,
					AccessPoint: conn.Client,
					HostSigner:  conn.ServerIdentity.KeySigner,
					Cluster:     conn.ServerIdentity.Cert.Extensions[utils.CertExtensionAuthority],
					Server:      s,
				})
			if err != nil {
				return trace.Wrap(err)
			}

			err = agentPool.Start()
			if err != nil {
				return trace.Wrap(err)
			}
			log.Infof("Service is starting in tunnel mode.")

			// Broadcast that the node has started.
			process.BroadcastEvent(Event{Name: NodeSSHReady, Payload: nil})
		}

		// Block and wait while the node is running.
		s.Wait()
		if conn.UseTunnel() {
			agentPool.Wait()
		}

		log.Infof("Exited.")
		return nil
	})

	// Execute this when process is asked to exit.
	process.OnExit("ssh.shutdown", func(payload interface{}) {
		if payload == nil {
			log.Infof("Shutting down immediately.")
			if s != nil {
				warnOnErr(s.Close(), log)
			}
		} else {
			log.Infof("Shutting down gracefully.")
			if s != nil {
				warnOnErr(s.Shutdown(payloadContext(payload, log)), log)
			}
		}
		if conn != nil && conn.UseTunnel() {
			agentPool.Stop()
		}

		if ebpf != nil {
			// Close BPF service.
			warnOnErr(ebpf.Close(), log)
		}

		if asyncEmitter != nil {
			warnOnErr(asyncEmitter.Close(), log)
		}

		log.Infof("Exited.")
	})

	return nil
}

// registerWithAuthServer uses one time provisioning token obtained earlier
// from the server to get a pair of SSH keys signed by Auth server host
// certificate authority
func (process *TeleportProcess) registerWithAuthServer(role types.SystemRole, eventName string) {
	serviceName := strings.ToLower(role.String())
	process.RegisterCriticalFunc(fmt.Sprintf("register.%v", serviceName), func() error {
		connector, err := process.reconnectToAuthService(role)
		if err != nil {
			return trace.Wrap(err)
		}
		process.OnExit(fmt.Sprintf("auth.client.%v", serviceName), func(interface{}) {
			process.log.Debugf("Closed client for %v.", role)
			err := connector.Client.Close()
			if err != nil {
				process.log.Debugf("Failed to close client: %v", err)
			}
		})
		process.BroadcastEvent(Event{Name: eventName, Payload: connector})
		return nil
	})
}

func (process *TeleportProcess) initUploaderService(accessPoint auth.AccessPoint, auditLog events.IAuditLog) error {
	log := process.log.WithFields(logrus.Fields{
		trace.Component: teleport.Component(teleport.ComponentAuditLog, process.id),
	})
	// create folder for uploads
	uid, gid, err := adminCreds()
	if err != nil {
		return trace.Wrap(err)
	}
	// prepare dirs for uploader
	streamingDir := []string{process.Config.DataDir, teleport.LogsDir, teleport.ComponentUpload, events.StreamingLogsDir, apidefaults.Namespace}
	paths := [][]string{
		// DELETE IN (5.1.0)
		// this directory will no longer be used after migration to 5.1.0
		{process.Config.DataDir, teleport.LogsDir, teleport.ComponentUpload, events.SessionLogsDir, apidefaults.Namespace},
		// This directory will remain to be used after migration to 5.1.0
		streamingDir,
	}
	for _, path := range paths {
		for i := 1; i < len(path); i++ {
			dir := filepath.Join(path[:i+1]...)
			log.Infof("Creating directory %v.", dir)
			err := os.Mkdir(dir, 0755)
			err = trace.ConvertSystemError(err)
			if err != nil {
				if !trace.IsAlreadyExists(err) {
					return trace.Wrap(err)
				}
			}
			if uid != nil && gid != nil {
				log.Infof("Setting directory %v owner to %v:%v.", dir, *uid, *gid)
				err := os.Chown(dir, *uid, *gid)
				if err != nil {
					return trace.ConvertSystemError(err)
				}
			}
		}
	}

	// DELETE IN (5.1.0)
	// this uploader was superseded by filesessions.Uploader,
	// see below
	uploader, err := events.NewUploader(events.UploaderConfig{
		DataDir:   filepath.Join(process.Config.DataDir, teleport.LogsDir),
		Namespace: apidefaults.Namespace,
		ServerID:  teleport.ComponentUpload,
		AuditLog:  auditLog,
		EventsC:   process.Config.UploadEventsC,
	})
	if err != nil {
		return trace.Wrap(err)
	}
	process.RegisterFunc("uploader.service", func() error {
		err := uploader.Serve()
		if err != nil {
			log.Errorf("Uploader server exited with error: %v.", err)
		}
		return nil
	})

	process.OnExit("uploader.shutdown", func(payload interface{}) {
		log.Infof("Shutting down.")
		warnOnErr(uploader.Stop(), log)
		log.Infof("Exited.")
	})

	// This uploader supersedes the events.Uploader above,
	// that is kept for backwards compatibility purposes for one release.
	// Delete this comment once the uploader above is phased out.
	fileUploader, err := filesessions.NewUploader(filesessions.UploaderConfig{
		ScanDir:  filepath.Join(streamingDir...),
		Streamer: accessPoint,
		AuditLog: auditLog,
		EventsC:  process.Config.UploadEventsC,
	})
	if err != nil {
		return trace.Wrap(err)
	}
	process.RegisterFunc("fileuploader.service", func() error {
		err := fileUploader.Serve()
		if err != nil {
			log.WithError(err).Errorf("File uploader server exited with error.")
		}
		return nil
	})

	process.OnExit("fileuploader.shutdown", func(payload interface{}) {
		log.Infof("File uploader is shutting down.")
		warnOnErr(fileUploader.Close(), log)
		log.Infof("File uploader has shut down.")
	})

	return nil
}

// initDiagnosticService starts diagnostic service currently serving healthz
// and prometheus endpoints
func (process *TeleportProcess) initDiagnosticService() error {
	mux := http.NewServeMux()
	mux.Handle("/metrics", promhttp.Handler())

	if process.Config.Debug {
		process.log.Infof("Adding diagnostic debugging handlers. To connect with profiler, use `go tool pprof %v`.", process.Config.DiagnosticAddr.Addr)

		mux.HandleFunc("/debug/pprof/", pprof.Index)
		mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
		mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
		mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
		mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
	}

	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
		roundtrip.ReplyJSON(w, http.StatusOK, map[string]interface{}{"status": "ok"})
	})

	log := process.log.WithFields(logrus.Fields{
		trace.Component: teleport.Component(teleport.ComponentDiagnostic, process.id),
	})

	// Create a state machine that will process and update the internal state of
	// Teleport based off Events. Use this state machine to return return the
	// status from the /readyz endpoint.
	ps, err := newProcessState(process)
	if err != nil {
		return trace.Wrap(err)
	}

	process.RegisterFunc("readyz.monitor", func() error {
		// Start loop to monitor for events that are used to update Teleport state.
		eventCh := make(chan Event, 1024)
		process.WaitForEvent(process.ExitContext(), TeleportDegradedEvent, eventCh)
		process.WaitForEvent(process.ExitContext(), TeleportOKEvent, eventCh)

		for {
			select {
			case e := <-eventCh:
				ps.update(e)
			case <-process.ExitContext().Done():
				log.Debugf("Teleport is exiting, returning.")
				return nil
			}
		}
	})
	mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) {
		switch ps.getState() {
		// 503
		case stateDegraded:
			roundtrip.ReplyJSON(w, http.StatusServiceUnavailable, map[string]interface{}{
				"status": "teleport is in a degraded state, check logs for details",
			})
		// 400
		case stateRecovering:
			roundtrip.ReplyJSON(w, http.StatusBadRequest, map[string]interface{}{
				"status": "teleport is recovering from a degraded state, check logs for details",
			})
		case stateStarting:
			roundtrip.ReplyJSON(w, http.StatusBadRequest, map[string]interface{}{
				"status": "teleport is starting and hasn't joined the cluster yet",
			})
		// 200
		case stateOK:
			roundtrip.ReplyJSON(w, http.StatusOK, map[string]interface{}{
				"status": "ok",
			})
		}
	})

	listener, err := process.importOrCreateListener(listenerDiagnostic, process.Config.DiagnosticAddr.Addr)
	if err != nil {
		return trace.Wrap(err)
	}
	warnOnErr(process.closeImportedDescriptors(teleport.ComponentDiagnostic), log)

	server := &http.Server{
		Handler:           mux,
		ReadHeaderTimeout: apidefaults.DefaultDialTimeout,
		ErrorLog:          utils.NewStdlogger(log.Error, teleport.ComponentDiagnostic),
	}

	log.Infof("Starting diagnostic service on %v.", process.Config.DiagnosticAddr.Addr)

	process.RegisterFunc("diagnostic.service", func() error {
		err := server.Serve(listener)
		if err != nil && err != http.ErrServerClosed {
			log.Warningf("Diagnostic server exited with error: %v.", err)
		}
		return nil
	})

	process.OnExit("diagnostic.shutdown", func(payload interface{}) {
		if payload == nil {
			log.Infof("Shutting down immediately.")
			warnOnErr(server.Close(), log)
		} else {
			log.Infof("Shutting down gracefully.")
			ctx := payloadContext(payload, log)
			warnOnErr(server.Shutdown(ctx), log)
		}
		log.Infof("Exited.")
	})
	return nil
}

// getAdditionalPrincipals returns a list of additional principals to add
// to role's service certificates.
func (process *TeleportProcess) getAdditionalPrincipals(role types.SystemRole) ([]string, []string, error) {
	var principals []string
	var dnsNames []string
	if process.Config.Hostname != "" {
		principals = append(principals, process.Config.Hostname)
	}
	var addrs []utils.NetAddr
	switch role {
	case types.RoleProxy:
		addrs = append(process.Config.Proxy.PublicAddrs,
			utils.NetAddr{Addr: string(teleport.PrincipalLocalhost)},
			utils.NetAddr{Addr: string(teleport.PrincipalLoopbackV4)},
			utils.NetAddr{Addr: string(teleport.PrincipalLoopbackV6)},
			utils.NetAddr{Addr: reversetunnel.LocalKubernetes},
		)
		addrs = append(addrs, process.Config.Proxy.SSHPublicAddrs...)
		addrs = append(addrs, process.Config.Proxy.TunnelPublicAddrs...)
		addrs = append(addrs, process.Config.Proxy.PostgresPublicAddrs...)
		addrs = append(addrs, process.Config.Proxy.MySQLPublicAddrs...)
		addrs = append(addrs, process.Config.Proxy.Kube.PublicAddrs...)
		// Automatically add wildcards for every proxy public address for k8s SNI routing
		if process.Config.Proxy.Kube.Enabled {
			for _, publicAddr := range utils.JoinAddrSlices(process.Config.Proxy.PublicAddrs, process.Config.Proxy.Kube.PublicAddrs) {
				host, err := utils.Host(publicAddr.Addr)
				if err != nil {
					return nil, nil, trace.Wrap(err)
				}
				if ip := net.ParseIP(host); ip == nil {
					dnsNames = append(dnsNames, "*."+host)
				}
			}
		}
	case types.RoleAuth, types.RoleAdmin:
		addrs = process.Config.Auth.PublicAddrs
	case types.RoleNode:
		// DELETE IN 5.0: We are manually adding HostUUID here in order
		// to allow UUID based routing to function with older Auth Servers
		// which don't automatically add UUID to the principal list.
		principals = append(principals, process.Config.HostUUID)
		addrs = process.Config.SSH.PublicAddrs
		// If advertise IP is set, add it to the list of principals. Otherwise
		// add in the default (0.0.0.0) which will be replaced by the Auth Server
		// when a host certificate is issued.
		if process.Config.AdvertiseIP != "" {
			advertiseIP, err := utils.ParseAddr(process.Config.AdvertiseIP)
			if err != nil {
				return nil, nil, trace.Wrap(err)
			}
			addrs = append(addrs, *advertiseIP)
		} else {
			addrs = append(addrs, process.Config.SSH.Addr)
		}
	case types.RoleKube:
		addrs = append(addrs,
			utils.NetAddr{Addr: string(teleport.PrincipalLocalhost)},
			utils.NetAddr{Addr: string(teleport.PrincipalLoopbackV4)},
			utils.NetAddr{Addr: string(teleport.PrincipalLoopbackV6)},
			utils.NetAddr{Addr: reversetunnel.LocalKubernetes},
		)
		addrs = append(addrs, process.Config.Kube.PublicAddrs...)
	case types.RoleApp:
		principals = append(principals, process.Config.HostUUID)
	}
	for _, addr := range addrs {
		if addr.IsEmpty() {
			continue
		}
		host, err := utils.Host(addr.Addr)
		if err != nil {
			return nil, nil, trace.Wrap(err)
		}
		principals = append(principals, host)
	}
	return principals, dnsNames, nil
}

// initProxy gets called if teleport runs with 'proxy' role enabled.
// this means it will do two things:
//    1. serve a web UI
//    2. proxy SSH connections to nodes running with 'node' role
//    3. take care of reverse tunnels
func (process *TeleportProcess) initProxy() error {
	// If no TLS key was provided for the web UI, generate a self-signed cert
	if len(process.Config.Proxy.KeyPairs) == 0 &&
		!process.Config.Proxy.DisableTLS &&
		!process.Config.Proxy.DisableWebService &&
		!process.Config.Proxy.ACME.Enabled {
		err := initSelfSignedHTTPSCert(process.Config)
		if err != nil {
			return trace.Wrap(err)
		}
	}
	process.registerWithAuthServer(types.RoleProxy, ProxyIdentityEvent)
	process.RegisterCriticalFunc("proxy.init", func() error {
		eventsC := make(chan Event)
		process.WaitForEvent(process.ExitContext(), ProxyIdentityEvent, eventsC)

		var event Event
		select {
		case event = <-eventsC:
			process.log.Debugf("Received event %q.", event.Name)
		case <-process.ExitContext().Done():
			process.log.Debugf("Process is exiting.")
			return nil
		}

		conn, ok := (event.Payload).(*Connector)
		if !ok {
			return trace.BadParameter("unsupported connector type: %T", event.Payload)
		}

		err := process.initProxyEndpoint(conn)
		if err != nil {
			if conn.Client != nil {
				warnOnErr(conn.Client.Close(), process.log)
			}
			return trace.Wrap(err)
		}

		return nil
	})
	return nil
}

type proxyListeners struct {
	mux           *multiplexer.Mux
	tls           *multiplexer.WebListener
	ssh           net.Listener
	web           net.Listener
	reverseTunnel net.Listener
	kube          net.Listener
	db            dbListeners
}

// dbListeners groups database access listeners.
type dbListeners struct {
	// postgres serves Postgres clients.
	postgres net.Listener
	// mysql serves MySQL clients.
	mysql net.Listener
	// tls serves database clients that use plain TLS handshake.
	tls net.Listener
}

// Empty returns true if no database access listeners are initialized.
func (l *dbListeners) Empty() bool {
	return l.postgres == nil && l.mysql == nil && l.tls == nil
}

// Close closes all database access listeners.
func (l *dbListeners) Close() {
	if l.postgres != nil {
		l.postgres.Close()
	}
	if l.mysql != nil {
		l.mysql.Close()
	}
	if l.tls != nil {
		l.tls.Close()
	}
}

// Close closes all proxy listeners.
func (l *proxyListeners) Close() {
	if l.mux != nil {
		l.mux.Close()
	}
	if l.tls != nil {
		l.tls.Close()
	}
	if l.web != nil {
		l.web.Close()
	}
	if l.reverseTunnel != nil {
		l.reverseTunnel.Close()
	}
	if l.kube != nil {
		l.kube.Close()
	}
	if !l.db.Empty() {
		l.db.Close()
	}
}

// setupProxyListeners sets up web proxy listeners based on the configuration
func (process *TeleportProcess) setupProxyListeners() (*proxyListeners, error) {
	cfg := process.Config
	process.log.Debugf("Setup Proxy: Web Proxy Address: %v, Reverse Tunnel Proxy Address: %v", cfg.Proxy.WebAddr.Addr, cfg.Proxy.ReverseTunnelListenAddr.Addr)
	var err error
	var listeners proxyListeners

	listeners.ssh, err = process.importOrCreateListener(listenerProxySSH, cfg.Proxy.SSHAddr.Addr)
	if err != nil {
		return nil, trace.Wrap(err)
	}

	if cfg.Proxy.Kube.Enabled {
		process.log.Debugf("Setup Proxy: turning on Kubernetes proxy.")
		listener, err := process.importOrCreateListener(listenerProxyKube, cfg.Proxy.Kube.ListenAddr.Addr)
		if err != nil {
			return nil, trace.Wrap(err)
		}
		listeners.kube = listener
	}

	if !cfg.Proxy.MySQLAddr.IsEmpty() {
		process.log.Debugf("Setup Proxy: MySQL proxy address: %v.", cfg.Proxy.MySQLAddr.Addr)
		listener, err := process.importOrCreateListener(listenerProxyMySQL, cfg.Proxy.MySQLAddr.Addr)
		if err != nil {
			return nil, trace.Wrap(err)
		}
		listeners.db.mysql = listener
	}

	switch {
	case cfg.Proxy.DisableWebService && cfg.Proxy.DisableReverseTunnel:
		process.log.Debugf("Setup Proxy: Reverse tunnel proxy and web proxy are disabled.")
		return &listeners, nil
	case cfg.Proxy.ReverseTunnelListenAddr == cfg.Proxy.WebAddr && !cfg.Proxy.DisableTLS:
		process.log.Debugf("Setup Proxy: Reverse tunnel proxy and web proxy listen on the same port, multiplexing is on.")
		listener, err := process.importOrCreateListener(listenerProxyTunnelAndWeb, cfg.Proxy.WebAddr.Addr)
		if err != nil {
			return nil, trace.Wrap(err)
		}
		listeners.mux, err = multiplexer.New(multiplexer.Config{
			EnableProxyProtocol: cfg.Proxy.EnableProxyProtocol,
			Listener:            listener,
			DisableTLS:          cfg.Proxy.DisableWebService,
			DisableSSH:          cfg.Proxy.DisableReverseTunnel,
			DisableDB:           cfg.Proxy.DisableDatabaseProxy,
			ID:                  teleport.Component(teleport.ComponentProxy, "tunnel", "web", process.id),
		})
		if err != nil {
			listener.Close()
			return nil, trace.Wrap(err)
		}
		listeners.web = listeners.mux.TLS()
		listeners.db.postgres = listeners.mux.DB()
		listeners.reverseTunnel = listeners.mux.SSH()
		go listeners.mux.Serve()
		return &listeners, nil
	case cfg.Proxy.EnableProxyProtocol && !cfg.Proxy.DisableWebService && !cfg.Proxy.DisableTLS:
		process.log.Debugf("Setup Proxy: Proxy protocol is enabled for web service, multiplexing is on.")
		listener, err := process.importOrCreateListener(listenerProxyWeb, cfg.Proxy.WebAddr.Addr)
		if err != nil {
			return nil, trace.Wrap(err)
		}
		listeners.mux, err = multiplexer.New(multiplexer.Config{
			EnableProxyProtocol: cfg.Proxy.EnableProxyProtocol,
			Listener:            listener,
			DisableTLS:          false,
			DisableSSH:          true,
			DisableDB:           cfg.Proxy.DisableDatabaseProxy,
			ID:                  teleport.Component(teleport.ComponentProxy, "web", process.id),
		})
		if err != nil {
			listener.Close()
			return nil, trace.Wrap(err)
		}
		listeners.web = listeners.mux.TLS()
		listeners.db.postgres = listeners.mux.DB()
		listeners.reverseTunnel, err = process.importOrCreateListener(listenerProxyTunnel, cfg.Proxy.ReverseTunnelListenAddr.Addr)
		if err != nil {
			listener.Close()
			listeners.Close()
			return nil, trace.Wrap(err)
		}
		go listeners.mux.Serve()
		return &listeners, nil
	default:
		process.log.Debug("Setup Proxy: Proxy and reverse tunnel are listening on separate ports.")
		if !cfg.Proxy.DisableReverseTunnel {
			listeners.reverseTunnel, err = process.importOrCreateListener(listenerProxyTunnel, cfg.Proxy.ReverseTunnelListenAddr.Addr)
			if err != nil {
				listeners.Close()
				return nil, trace.Wrap(err)
			}
		}
		if !cfg.Proxy.DisableWebService {
			listener, err := process.importOrCreateListener(listenerProxyWeb, cfg.Proxy.WebAddr.Addr)
			if err != nil {
				listeners.Close()
				return nil, trace.Wrap(err)
			}
			// Unless database proxy is explicitly disabled (which is currently
			// only done by tests and not exposed via file config), the web
			// listener is multiplexing both web and db client connections.
			if !cfg.Proxy.DisableDatabaseProxy && !cfg.Proxy.DisableTLS {
				process.log.Debug("Setup Proxy: Multiplexing web and database proxy on the same port.")
				listeners.mux, err = multiplexer.New(multiplexer.Config{
					EnableProxyProtocol: cfg.Proxy.EnableProxyProtocol,
					Listener:            listener,
					DisableTLS:          false, // Web service is enabled or we wouldn't have gotten here.
					DisableSSH:          true,  // Reverse tunnel is on a separate listener created above.
					DisableDB:           false, // Database proxy is enabled or we wouldn't have gotten here.
					ID:                  teleport.Component(teleport.ComponentProxy, "web", process.id),
				})
				if err != nil {
					listener.Close()
					listeners.Close()
					return nil, trace.Wrap(err)
				}
				listeners.web = listeners.mux.TLS()
				listeners.db.postgres = listeners.mux.DB()
				go listeners.mux.Serve()
			} else {
				process.log.Debug("Setup Proxy: TLS is disabled, multiplexing is off.")
				listeners.web = listener
			}
		}
		return &listeners, nil
	}
}

func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error {
	// clean up unused descriptors passed for proxy, but not used by it
	defer func() {
		if err := process.closeImportedDescriptors(teleport.ComponentProxy); err != nil {
			process.log.Warnf("Failed closing imported file descriptors: %v", err)
		}
	}()
	var err error
	cfg := process.Config

	proxyLimiter, err := limiter.NewLimiter(cfg.Proxy.Limiter)
	if err != nil {
		return trace.Wrap(err)
	}

	reverseTunnelLimiter, err := limiter.NewLimiter(cfg.Proxy.Limiter)
	if err != nil {
		return trace.Wrap(err)
	}

	// make a caching auth client for the auth server:
	accessPoint, err := process.newLocalCacheForProxy(conn.Client, []string{teleport.ComponentProxy})
	if err != nil {
		return trace.Wrap(err)
	}

	clientTLSConfig, err := conn.ClientIdentity.TLSConfig(cfg.CipherSuites)
	if err != nil {
		return trace.Wrap(err)
	}

	listeners, err := process.setupProxyListeners()
	if err != nil {
		return trace.Wrap(err)
	}

	proxySSHAddr := cfg.Proxy.SSHAddr
	// override value of cfg.Proxy.SSHAddr with listener addr in order
	// to support binding to a random port (e.g. `127.0.0.1:0`).
	proxySSHAddr.Addr = listeners.ssh.Addr().String()

	log := process.log.WithFields(logrus.Fields{
		trace.Component: teleport.Component(teleport.ComponentReverseTunnelServer, process.id),
	})

	clusterName := conn.ServerIdentity.Cert.Extensions[utils.CertExtensionAuthority]

	// asyncEmitter makes sure that sessions do not block
	// in case if connections are slow
	asyncEmitter, err := process.newAsyncEmitter(conn.Client)
	if err != nil {
		return trace.Wrap(err)
	}
	streamer, err := events.NewCheckingStreamer(events.CheckingStreamerConfig{
		Inner:       conn.Client,
		Clock:       process.Clock,
		ClusterName: clusterName,
	})
	if err != nil {
		return trace.Wrap(err)
	}
	streamEmitter := &events.StreamerAndEmitter{
		Emitter:  asyncEmitter,
		Streamer: streamer,
	}

	// register SSH reverse tunnel server that accepts connections
	// from remote teleport nodes
	var tsrv reversetunnel.Server
	if !process.Config.Proxy.DisableReverseTunnel {
		tsrv, err = reversetunnel.NewServer(
			reversetunnel.Config{
				Component:                     teleport.Component(teleport.ComponentProxy, process.id),
				ID:                            process.Config.HostUUID,
				ClusterName:                   clusterName,
				ClientTLS:                     clientTLSConfig,
				Listener:                      listeners.reverseTunnel,
				HostSigners:                   []ssh.Signer{conn.ServerIdentity.KeySigner},
				LocalAuthClient:               conn.Client,
				LocalAccessPoint:              accessPoint,
				NewCachingAccessPoint:         process.newLocalCacheForRemoteProxy,
				NewCachingAccessPointOldProxy: process.newLocalCacheForOldRemoteProxy,
				Limiter:                       reverseTunnelLimiter,
				DirectClusters: []reversetunnel.DirectCluster{
					{
						Name:   conn.ServerIdentity.Cert.Extensions[utils.CertExtensionAuthority],
						Client: conn.Client,
					},
				},
				KeyGen:        cfg.Keygen,
				Ciphers:       cfg.Ciphers,
				KEXAlgorithms: cfg.KEXAlgorithms,
				MACAlgorithms: cfg.MACAlgorithms,
				DataDir:       process.Config.DataDir,
				PollingPeriod: process.Config.PollingPeriod,
				FIPS:          cfg.FIPS,
				Emitter:       streamEmitter,
				Log:           process.log,
			})
		if err != nil {
			return trace.Wrap(err)
		}
		process.RegisterCriticalFunc("proxy.reversetunnel.server", func() error {
			utils.Consolef(cfg.Console, log, teleport.ComponentProxy, "Reverse tunnel service %s:%s is starting on %v.",
				teleport.Version, teleport.Gitref, cfg.Proxy.ReverseTunnelListenAddr.Addr)
			log.Infof("Starting %s:%s on %v using %v", teleport.Version, teleport.Gitref, cfg.Proxy.ReverseTunnelListenAddr.Addr, process.Config.CachePolicy)
			if err := tsrv.Start(); err != nil {
				log.Error(err)
				return trace.Wrap(err)
			}

			// notify parties that we've started reverse tunnel server
			process.BroadcastEvent(Event{Name: ProxyReverseTunnelReady, Payload: tsrv})
			tsrv.Wait()
			return nil
		})
	}

	// Register web proxy server
	var webServer *http.Server
	var webHandler *web.RewritingHandler
	if !process.Config.Proxy.DisableWebService {
		proxySettings := webclient.ProxySettings{
			Kube: webclient.KubeProxySettings{
				Enabled: cfg.Proxy.Kube.Enabled,
			},
			SSH: webclient.SSHProxySettings{
				ListenAddr:       proxySSHAddr.Addr,
				TunnelListenAddr: cfg.Proxy.ReverseTunnelListenAddr.String(),
			},
		}
		if len(cfg.Proxy.PublicAddrs) > 0 {
			proxySettings.SSH.PublicAddr = cfg.Proxy.PublicAddrs[0].String()
		}
		if len(cfg.Proxy.SSHPublicAddrs) > 0 {
			proxySettings.SSH.SSHPublicAddr = cfg.Proxy.SSHPublicAddrs[0].String()
		}
		if len(cfg.Proxy.TunnelPublicAddrs) > 0 {
			proxySettings.SSH.TunnelPublicAddr = cfg.Proxy.TunnelPublicAddrs[0].String()
		}
		if cfg.Proxy.Kube.Enabled {
			proxySettings.Kube.ListenAddr = cfg.Proxy.Kube.ListenAddr.String()
		}
		if len(cfg.Proxy.Kube.PublicAddrs) > 0 {
			proxySettings.Kube.PublicAddr = cfg.Proxy.Kube.PublicAddrs[0].String()
		}
		if len(cfg.Proxy.PostgresPublicAddrs) > 0 {
			proxySettings.DB.PostgresPublicAddr = cfg.Proxy.PostgresPublicAddrs[0].String()
		}
		if !cfg.Proxy.MySQLAddr.IsEmpty() {
			proxySettings.DB.MySQLListenAddr = cfg.Proxy.MySQLAddr.String()
		}
		if len(cfg.Proxy.MySQLPublicAddrs) > 0 {
			proxySettings.DB.MySQLPublicAddr = cfg.Proxy.MySQLPublicAddrs[0].String()
		}
		var fs http.FileSystem
		if !process.Config.Proxy.DisableWebInterface {
			fs, err = newHTTPFileSystem()
			if err != nil {
				return trace.Wrap(err)
			}
		}

		webHandler, err = web.NewHandler(
			web.Config{
				Proxy:            tsrv,
				AuthServers:      cfg.AuthServers[0],
				DomainName:       cfg.Hostname,
				ProxyClient:      conn.Client,
				ProxySSHAddr:     proxySSHAddr,
				ProxyWebAddr:     cfg.Proxy.WebAddr,
				ProxyPublicAddrs: cfg.Proxy.PublicAddrs,
				ProxySettings:    proxySettings,
				CipherSuites:     cfg.CipherSuites,
				FIPS:             cfg.FIPS,
				AccessPoint:      accessPoint,
				Emitter:          streamEmitter,
				PluginRegistry:   process.PluginRegistry,
				HostUUID:         process.Config.HostUUID,
				Context:          process.ExitContext(),
				StaticFS:         fs,
				ClusterFeatures:  process.getClusterFeatures(),
			})
		if err != nil {
			return trace.Wrap(err)
		}
		proxyLimiter.WrapHandle(webHandler)
		if !process.Config.Proxy.DisableTLS {
			var tlsConfig *tls.Config
			acmeCfg := process.Config.Proxy.ACME
			if !acmeCfg.Enabled {
				tlsConfig = utils.TLSConfig(cfg.CipherSuites)
			} else {
				log.Infof("Managing certs using ACME https://datatracker.ietf.org/doc/rfc8555/.")

				acmePath := filepath.Join(process.Config.DataDir, teleport.ComponentACME)
				if err := os.MkdirAll(acmePath, teleport.PrivateDirMode); err != nil {
					return trace.ConvertSystemError(err)
				}
				hostChecker, err := newHostPolicyChecker(hostPolicyCheckerConfig{
					publicAddrs: process.Config.Proxy.PublicAddrs,
					clt:         conn.Client,
					tun:         tsrv,
					clusterName: conn.ServerIdentity.Cert.Extensions[utils.CertExtensionAuthority],
				})
				if err != nil {
					return trace.Wrap(err)
				}
				m := &autocert.Manager{
					Cache:      autocert.DirCache(acmePath),
					Prompt:     autocert.AcceptTOS,
					HostPolicy: hostChecker.checkHost,
					Email:      acmeCfg.Email,
				}
				if acmeCfg.URI != "" {
					m.Client = &acme.Client{DirectoryURL: acmeCfg.URI}
				}
				tlsConfig = m.TLSConfig()
				utils.SetupTLSConfig(tlsConfig, cfg.CipherSuites)
			}

			for _, pair := range process.Config.Proxy.KeyPairs {
				log.Infof("Loading TLS certificate %v and key %v.", pair.Certificate, pair.PrivateKey)

				certificate, err := tls.LoadX509KeyPair(pair.Certificate, pair.PrivateKey)
				if err != nil {
					return trace.Wrap(err)
				}
				tlsConfig.Certificates = append(tlsConfig.Certificates, certificate)
			}

			tlsConfig.GetConfigForClient = func(*tls.ClientHelloInfo) (*tls.Config, error) {
				tlsClone := tlsConfig.Clone()

				// Set client auth to "verify client cert if given" to support
				// app access CLI flow.
				//
				// Clients (like curl) connecting to the web proxy endpoint will
				// present a client certificate signed by the cluster's user CA.
				//
				// Browser connections to web UI and other clients (like database
				// access) connecting to web proxy won't be affected since they
				// don't present a certificate.
				tlsClone.ClientAuth = tls.VerifyClientCertIfGiven

				// Build the client CA pool containing the cluster's user CA in
				// order to be able to validate certificates provided by app
				// access CLI clients.
				tlsClone.ClientCAs, err = auth.ClientCertPool(accessPoint, clusterName)
				if err != nil {
					return nil, trace.Wrap(err)
				}

				return tlsClone, nil
			}

			// Setup a listener that will multiplex TLS connections between
			// those that will be served by a web server (such as connections
			// to web UI or application access) and those served by a database
			// access for databases that use plain TLS handshake (such as
			// MongoDB).
			listeners.tls, err = multiplexer.NewWebListener(multiplexer.WebListenerConfig{
				Listener: tls.NewListener(listeners.web, tlsConfig),
			})
			if err != nil {
				return trace.Wrap(err)
			}
			listeners.web = listeners.tls.Web()
			listeners.db.tls = listeners.tls.DB()

			process.RegisterCriticalFunc("proxy.tls", func() error {
				log.Infof("TLS multiplexer is starting on %v.", cfg.Proxy.WebAddr.Addr)
				if err := listeners.tls.Serve(); !trace.IsConnectionProblem(err) {
					log.WithError(err).Warn("TLS multiplexer error.")
				}
				log.Info("TLS multiplexer exited.")
				return nil
			})
		}
		webServer = &http.Server{
			Handler:           proxyLimiter,
			ReadHeaderTimeout: apidefaults.DefaultDialTimeout,
			ErrorLog:          utils.NewStdlogger(log.Error, teleport.ComponentProxy),
		}
		process.RegisterCriticalFunc("proxy.web", func() error {
			utils.Consolef(cfg.Console, log, teleport.ComponentProxy, "Web proxy service %s:%s is starting on %v.",
				teleport.Version, teleport.Gitref, cfg.Proxy.WebAddr.Addr)
			log.Infof("Web proxy service %s:%s is starting on %v.", teleport.Version, teleport.Gitref, cfg.Proxy.WebAddr.Addr)
			defer webHandler.Close()
			process.BroadcastEvent(Event{Name: ProxyWebServerReady, Payload: webHandler})
			if err := webServer.Serve(listeners.web); err != nil && err != http.ErrServerClosed {
				log.Warningf("Error while serving web requests: %v", err)
			}
			log.Info("Exited.")
			return nil
		})
	} else {
		log.Info("Web UI is disabled.")
	}

	sshProxy, err := regular.New(proxySSHAddr,
		cfg.Hostname,
		[]ssh.Signer{conn.ServerIdentity.KeySigner},
		accessPoint,
		cfg.DataDir,
		"",
		process.proxyPublicAddr(),
		regular.SetLimiter(proxyLimiter),
		regular.SetProxyMode(tsrv),
		regular.SetSessionServer(conn.Client),
		regular.SetCiphers(cfg.Ciphers),
		regular.SetKEXAlgorithms(cfg.KEXAlgorithms),
		regular.SetMACAlgorithms(cfg.MACAlgorithms),
		regular.SetNamespace(apidefaults.Namespace),
		regular.SetRotationGetter(process.getRotation),
		regular.SetFIPS(cfg.FIPS),
		regular.SetOnHeartbeat(func(err error) {
			if err != nil {
				process.BroadcastEvent(Event{Name: TeleportDegradedEvent, Payload: teleport.ComponentProxy})
			} else {
				process.BroadcastEvent(Event{Name: TeleportOKEvent, Payload: teleport.ComponentProxy})
			}
		}),
		regular.SetEmitter(streamEmitter),
	)
	if err != nil {
		return trace.Wrap(err)
	}

	process.RegisterCriticalFunc("proxy.ssh", func() error {
		utils.Consolef(cfg.Console, log, teleport.ComponentProxy, "SSH proxy service %s:%s is starting on %v.",
			teleport.Version, teleport.Gitref, proxySSHAddr.Addr)
		log.Infof("SSH proxy service %s:%s is starting on %v", teleport.Version, teleport.Gitref, proxySSHAddr)
		go sshProxy.Serve(listeners.ssh)
		// broadcast that the proxy ssh server has started
		process.BroadcastEvent(Event{Name: ProxySSHReady, Payload: nil})
		return nil
	})

	// Create and register reverse tunnel AgentPool.
	rcWatcher, err := reversetunnel.NewRemoteClusterTunnelManager(reversetunnel.RemoteClusterTunnelManagerConfig{
		HostUUID:            conn.ServerIdentity.ID.HostUUID,
		AuthClient:          conn.Client,
		AccessPoint:         accessPoint,
		HostSigner:          conn.ServerIdentity.KeySigner,
		LocalCluster:        conn.ServerIdentity.Cert.Extensions[utils.CertExtensionAuthority],
		KubeDialAddr:        utils.DialAddrFromListenAddr(cfg.Proxy.Kube.ListenAddr),
		ReverseTunnelServer: tsrv,
	})
	if err != nil {
		return trace.Wrap(err)
	}

	process.RegisterCriticalFunc("proxy.reversetunnel.watcher", func() error {
		log := logrus.WithFields(logrus.Fields{
			trace.Component: teleport.Component(teleport.ComponentReverseTunnelAgent, process.id),
		})
		log.Infof("Starting reverse tunnel agent pool.")
		done := make(chan struct{})
		go func() {
			defer close(done)
			rcWatcher.Run(process.ExitContext())
		}()
		process.BroadcastEvent(Event{Name: ProxyAgentPoolReady, Payload: rcWatcher})
		<-done
		return nil
	})

	var kubeServer *kubeproxy.TLSServer
	if listeners.kube != nil && !process.Config.Proxy.DisableReverseTunnel {
		authorizer, err := auth.NewAuthorizer(clusterName, conn.Client, conn.Client, conn.Client)
		if err != nil {
			return trace.Wrap(err)
		}
		// Register TLS endpoint of the Kube proxy service
		tlsConfig, err := conn.ServerIdentity.TLSConfig(cfg.CipherSuites)
		if err != nil {
			return trace.Wrap(err)
		}
		component := teleport.Component(teleport.ComponentProxy, teleport.ComponentProxyKube)
		kubeServiceType := kubeproxy.ProxyService
		if cfg.Proxy.Kube.LegacyKubeProxy {
			kubeServiceType = kubeproxy.LegacyProxyService
		}
		kubeServer, err = kubeproxy.NewTLSServer(kubeproxy.TLSServerConfig{
			ForwarderConfig: kubeproxy.ForwarderConfig{
				Namespace:         apidefaults.Namespace,
				Keygen:            cfg.Keygen,
				ClusterName:       clusterName,
				ReverseTunnelSrv:  tsrv,
				Authz:             authorizer,
				AuthClient:        conn.Client,
				StreamEmitter:     streamEmitter,
				DataDir:           cfg.DataDir,
				CachingAuthClient: accessPoint,
				ServerID:          cfg.HostUUID,
				ClusterOverride:   cfg.Proxy.Kube.ClusterOverride,
				KubeconfigPath:    cfg.Proxy.Kube.KubeconfigPath,
				Component:         component,
				KubeServiceType:   kubeServiceType,
			},
			TLS:           tlsConfig,
			LimiterConfig: cfg.Proxy.Limiter,
			AccessPoint:   accessPoint,
			OnHeartbeat: func(err error) {
				if err != nil {
					process.BroadcastEvent(Event{Name: TeleportDegradedEvent, Payload: component})
				} else {
					process.BroadcastEvent(Event{Name: TeleportOKEvent, Payload: component})
				}
			},
		})
		if err != nil {
			return trace.Wrap(err)
		}
		process.RegisterCriticalFunc("proxy.kube", func() error {
			log := logrus.WithFields(logrus.Fields{
				trace.Component: component,
			})
			log.Infof("Starting Kube proxy on %v.", cfg.Proxy.Kube.ListenAddr.Addr)
			err := kubeServer.Serve(listeners.kube)
			if err != nil && err != http.ErrServerClosed {
				log.Warningf("Kube TLS server exited with error: %v.", err)
			}
			return nil
		})
	}

	// Start the database proxy server that will be accepting connections from
	// the database clients (such as psql or mysql), authenticating them, and
	// then routing them to a respective database server over the reverse tunnel
	// framework.
	if !listeners.db.Empty() && !process.Config.Proxy.DisableReverseTunnel {
		authorizer, err := auth.NewAuthorizer(clusterName, conn.Client, conn.Client, conn.Client)
		if err != nil {
			return trace.Wrap(err)
		}
		tlsConfig, err := conn.ServerIdentity.TLSConfig(cfg.CipherSuites)
		if err != nil {
			return trace.Wrap(err)
		}
		dbProxyServer, err := db.NewProxyServer(process.ExitContext(),
			db.ProxyServerConfig{
				AuthClient:  conn.Client,
				AccessPoint: accessPoint,
				Authorizer:  authorizer,
				Tunnel:      tsrv,
				TLSConfig:   tlsConfig,
				Emitter:     asyncEmitter,
				Clock:       process.Clock,
				ServerID:    cfg.HostUUID,
			})
		if err != nil {
			return trace.Wrap(err)
		}
		log := process.log.WithField(trace.Component, teleport.Component(teleport.ComponentDatabase))
		if listeners.db.postgres != nil {
			process.RegisterCriticalFunc("proxy.db.postgres", func() error {
				log.Infof("Starting Postgres proxy server on %v.", cfg.Proxy.WebAddr.Addr)
				if err := dbProxyServer.Serve(listeners.db.postgres); err != nil {
					log.WithError(err).Warn("Postgres proxy server exited with error.")
				}
				return nil
			})
		}
		if listeners.db.mysql != nil {
			process.RegisterCriticalFunc("proxy.db.mysql", func() error {
				log.Infof("Starting MySQL proxy server on %v.", cfg.Proxy.MySQLAddr.Addr)
				if err := dbProxyServer.ServeMySQL(listeners.db.mysql); err != nil {
					log.WithError(err).Warn("MySQL proxy server exited with error.")
				}
				return nil
			})
		}
		if listeners.db.tls != nil {
			process.RegisterCriticalFunc("proxy.db.tls", func() error {
				log.Infof("Starting Database TLS proxy server on %v.", cfg.Proxy.WebAddr.Addr)
				if err := dbProxyServer.ServeTLS(listeners.db.tls); err != nil {
					log.WithError(err).Warn("Database TLS proxy server exited with error.")
				}
				return nil
			})
		}
	}

	// execute this when process is asked to exit:
	process.OnExit("proxy.shutdown", func(payload interface{}) {
		rcWatcher.Close()
		defer listeners.Close()
		// Need to shut down this listener first, because
		// in case of graceful shutdown, if tls server was not called
		// the shutdown could be doing nothing, as server has not
		// started tracking the listener first. It's ok to close listener
		// several times.
		if listeners.kube != nil {
			listeners.kube.Close()
		}
		if asyncEmitter != nil {
			warnOnErr(asyncEmitter.Close(), log)
		}
		if payload == nil {
			log.Infof("Shutting down immediately.")
			if tsrv != nil {
				warnOnErr(tsrv.Close(), log)
			}
			if webServer != nil {
				warnOnErr(webServer.Close(), log)
			}
			if webHandler != nil {
				warnOnErr(webHandler.Close(), log)
			}
			warnOnErr(sshProxy.Close(), log)
			if kubeServer != nil {
				warnOnErr(kubeServer.Close(), log)
			}
		} else {
			log.Infof("Shutting down gracefully.")
			ctx := payloadContext(payload, log)
			if tsrv != nil {
				warnOnErr(tsrv.Shutdown(ctx), log)
			}
			warnOnErr(sshProxy.Shutdown(ctx), log)
			if webServer != nil {
				warnOnErr(webServer.Shutdown(ctx), log)
			}
			if kubeServer != nil {
				warnOnErr(kubeServer.Shutdown(ctx), log)
			}
			if webHandler != nil {
				warnOnErr(webHandler.Close(), log)
			}
		}
		// Close client after graceful shutdown has been completed,
		// to make sure in flight streams are not terminated,
		if conn.Client != nil {
			warnOnErr(conn.Client.Close(), log)
		}
		log.Infof("Exited.")
	})
	if err := process.initUploaderService(accessPoint, conn.Client); err != nil {
		return trace.Wrap(err)
	}
	return nil
}

// registerAppDepend will register dependencies for application service.
func (process *TeleportProcess) registerAppDepend() {
	for _, eventName := range appDependEvents {
		process.WaitForEvent(process.ExitContext(), eventName, process.appDependCh)
	}
}

// waitForAppDepend waits until all dependencies for an application service
// are ready.
func (process *TeleportProcess) waitForAppDepend() {
	for i := 0; i < len(appDependEvents); i++ {
		select {
		case <-process.appDependCh:
		case <-process.ExitContext().Done():
			process.log.Debugf("Process is exiting.")
		}
	}
}

// appDependEvents is a list of events that the application service depends on.
var appDependEvents = []string{
	AuthTLSReady,
	AuthIdentityEvent,
	ProxySSHReady,
	ProxyWebServerReady,
	ProxyReverseTunnelReady,
}

func (process *TeleportProcess) initApps() {
	// If no applications are specified, exit early. This is due to the strange
	// behavior in reading file configuration. If the user does not specify an
	// "app_service" section, that is considered enabling "app_service".
	if len(process.Config.Apps.Apps) == 0 && !process.Config.Apps.DebugApp {
		return
	}

	// Connect to the Auth Server, a client connected to the Auth Server will
	// be returned. For this to be successful, credentials to connect to the
	// Auth Server need to exist on disk or a registration token should be
	// provided.
	process.registerWithAuthServer(types.RoleApp, AppsIdentityEvent)
	eventsCh := make(chan Event)
	process.WaitForEvent(process.ExitContext(), AppsIdentityEvent, eventsCh)

	// Define logger to prefix log lines with the name of the component and PID.
	component := teleport.Component(teleport.ComponentApp, process.id)
	log := process.log.WithField(trace.Component, component)

	var appServer *app.Server
	var agentPool *reversetunnel.AgentPool

	process.RegisterCriticalFunc("apps.start", func() error {
		var ok bool
		var event Event

		// Block until registration is complete and a client (with an identity) has
		// been returned.
		select {
		case event = <-eventsCh:
			log.Debugf("Received event %q.", event.Name)
		case <-process.ExitContext().Done():
			log.Debugf("Process is exiting.")
			return nil
		}
		conn, ok := (event.Payload).(*Connector)
		if !ok {
			return trace.BadParameter("unsupported event payload type %q", event.Payload)
		}

		// If this process connected through the web proxy, it will discover the
		// reverse tunnel address correctly and store it in the connector.
		//
		// If it was not, it is running in single process mode which is used for
		// development and demos. In that case, wait until all dependencies (like
		// auth and reverse tunnel server) are ready before starting.
		var tunnelAddr string
		if conn.TunnelProxy() != "" {
			tunnelAddr = conn.TunnelProxy()
		} else {
			if tunnelAddr, ok = process.singleProcessMode(); !ok {
				return trace.BadParameter(`failed to find reverse tunnel address, if running in single process mode, make sure "auth_service", "proxy_service", and "app_service" are all enabled`)
			}

			// Block and wait for all dependencies to start before starting.
			log.Debugf("Waiting for application service dependencies to start.")
			process.waitForAppDepend()
		}

		// Create a caching client to the Auth Server. It is to reduce load on
		// the Auth Server.
		accessPoint, err := process.newLocalCache(conn.Client, cache.ForApps, []string{component})
		if err != nil {
			return trace.Wrap(err)
		}

		// Start uploader that will scan a path on disk and upload completed
		// sessions to the Auth Server.
		if err := process.initUploaderService(accessPoint, conn.Client); err != nil {
			return trace.Wrap(err)
		}

		// Start header dumping debugging application if requested.
		if process.Config.Apps.DebugApp {
			debugCh := make(chan Event)
			process.WaitForEvent(process.ExitContext(), DebugAppReady, debugCh)
			process.initDebugApp()

			// Block until the header dumper application is ready, and once it is,
			// figure out where it's running and add it to the list of applications.
			select {
			case event := <-debugCh:
				server, ok := event.Payload.(*httptest.Server)
				if !ok {
					return trace.BadParameter("unexpected payload %T", event.Payload)
				}
				process.Config.Apps.Apps = append(process.Config.Apps.Apps, App{
					Name: "dumper",
					URI:  server.URL,
				})
			case <-process.ExitContext().Done():
				return trace.Wrap(process.ExitContext().Err())
			}
		}

		// Loop over each application and create a server.
		var applications []*types.App
		for _, app := range process.Config.Apps.Apps {
			publicAddr, err := getPublicAddr(accessPoint, app)
			if err != nil {
				return trace.Wrap(err)
			}

			a := &types.App{
				Name:               app.Name,
				Description:        app.Description,
				URI:                app.URI,
				PublicAddr:         publicAddr,
				StaticLabels:       app.StaticLabels,
				DynamicLabels:      types.LabelsToV2(app.DynamicLabels),
				InsecureSkipVerify: app.InsecureSkipVerify,
			}
			if app.Rewrite != nil {
				a.Rewrite = &types.Rewrite{
					Redirect: app.Rewrite.Redirect,
				}
				for _, header := range app.Rewrite.Headers {
					a.Rewrite.Headers = append(a.Rewrite.Headers,
						&types.Header{
							Name:  header.Name,
							Value: header.Value,
						})
				}
			}

			applications = append(applications, a)
		}
		server := &types.ServerV2{
			Kind:    types.KindAppServer,
			Version: types.V2,
			Metadata: types.Metadata{
				Namespace: apidefaults.Namespace,
				Name:      process.Config.HostUUID,
			},
			Spec: types.ServerSpecV2{
				Hostname: process.Config.Hostname,
				Version:  teleport.Version,
				Apps:     applications,
			},
		}
		clusterName := conn.ServerIdentity.Cert.Extensions[utils.CertExtensionAuthority]

		authorizer, err := auth.NewAuthorizer(clusterName, conn.Client, conn.Client, conn.Client)
		if err != nil {
			return trace.Wrap(err)
		}
		tlsConfig, err := conn.ServerIdentity.TLSConfig(nil)
		if err != nil {
			return trace.Wrap(err)
		}

		appServer, err = app.New(process.ExitContext(), &app.Config{
			DataDir:      process.Config.DataDir,
			AuthClient:   conn.Client,
			AccessPoint:  accessPoint,
			Authorizer:   authorizer,
			TLSConfig:    tlsConfig,
			CipherSuites: process.Config.CipherSuites,
			GetRotation:  process.getRotation,
			Server:       server,
			OnHeartbeat: func(err error) {
				if err != nil {
					process.BroadcastEvent(Event{Name: TeleportDegradedEvent, Payload: teleport.ComponentApp})
				} else {
					process.BroadcastEvent(Event{Name: TeleportOKEvent, Payload: teleport.ComponentApp})
				}
			},
		})
		if err != nil {
			return trace.Wrap(err)
		}

		// Start the apps server. This starts the server, heartbeat (services.App),
		// and (dynamic) label update.
		appServer.Start()

		// Create and start an agent pool.
		agentPool, err = reversetunnel.NewAgentPool(process.ExitContext(),
			reversetunnel.AgentPoolConfig{
				Component:   teleport.ComponentApp,
				HostUUID:    conn.ServerIdentity.ID.HostUUID,
				ProxyAddr:   tunnelAddr,
				Client:      conn.Client,
				Server:      appServer,
				AccessPoint: accessPoint,
				HostSigner:  conn.ServerIdentity.KeySigner,
				Cluster:     clusterName,
			})
		if err != nil {
			return trace.Wrap(err)
		}
		err = agentPool.Start()
		if err != nil {
			return trace.Wrap(err)
		}

		process.BroadcastEvent(Event{Name: AppsReady, Payload: nil})
		log.Infof("All applications successfully started.")

		// Block and wait while the server and agent pool are running.
		if err := appServer.Wait(); err != nil {
			return trace.Wrap(err)
		}
		agentPool.Wait()

		return nil
	})

	// Execute this when process is asked to exit.
	process.OnExit("apps.stop", func(payload interface{}) {
		log.Infof("Shutting down.")
		if appServer != nil {
			warnOnErr(appServer.Close(), log)
		}
		if agentPool != nil {
			agentPool.Stop()
		}

		log.Infof("Exited.")
	})
}

func warnOnErr(err error, log logrus.FieldLogger) {
	if err != nil {
		// don't warn on double close, happens sometimes when
		// calling accept on a closed listener
		if strings.Contains(err.Error(), constants.UseOfClosedNetworkConnection) {
			return
		}
		log.WithError(err).Warn("Got error while cleaning up.")
	}
}

// initAuthStorage initializes the storage backend for the auth service.
func (process *TeleportProcess) initAuthStorage() (bk backend.Backend, err error) {
	ctx := context.TODO()
	bc := &process.Config.Auth.StorageConfig
	process.log.Debugf("Using %v backend.", bc.Type)
	switch bc.Type {
	// SQLite backend (or alt name dir).
	case lite.GetName():
		bk, err = lite.New(ctx, bc.Params)
	// Firestore backend:
	case firestore.GetName():
		bk, err = firestore.New(ctx, bc.Params)
	// DynamoDB backend.
	case dynamo.GetName():
		bk, err = dynamo.New(ctx, bc.Params)
	// etcd backend.
	case etcdbk.GetName():
		bk, err = etcdbk.New(ctx, bc.Params)
	default:
		err = trace.BadParameter("unsupported secrets storage type: %q", bc.Type)
	}
	if err != nil {
		return nil, trace.Wrap(err)
	}
	if err := bk.Migrate(ctx); err != nil {
		return nil, trace.Wrap(err)
	}
	reporter, err := backend.NewReporter(backend.ReporterConfig{
		Component: teleport.ComponentBackend,
		Backend:   backend.NewSanitizer(bk),
	})
	if err != nil {
		return nil, trace.Wrap(err)
	}
	process.setReporter(reporter)
	return reporter, nil
}

func (process *TeleportProcess) setReporter(reporter *backend.Reporter) {
	process.Lock()
	defer process.Unlock()
	process.reporter = reporter
}

// WaitWithContext waits until all internal services stop.
func (process *TeleportProcess) WaitWithContext(ctx context.Context) {
	local, cancel := context.WithCancel(ctx)
	go func() {
		defer cancel()
		if err := process.Supervisor.Wait(); err != nil {
			process.log.Warnf("Error waiting for all services to complete: %v", err)
		}
	}()

	<-local.Done()
}

// StartShutdown launches non-blocking graceful shutdown process that signals
// completion, returns context that will be closed once the shutdown is done
func (process *TeleportProcess) StartShutdown(ctx context.Context) context.Context {
	process.BroadcastEvent(Event{Name: TeleportExitEvent, Payload: ctx})
	localCtx, cancel := context.WithCancel(ctx)
	go func() {
		defer cancel()
		if err := process.Supervisor.Wait(); err != nil {
			process.log.Warnf("Error waiting for all services to complete: %v", err)
		}
		process.log.Debug("All supervisor functions are completed.")
		localAuth := process.getLocalAuth()
		if localAuth != nil {
			if err := process.localAuth.Close(); err != nil {
				process.log.Warningf("Failed closing auth server: %v.", err)
			}
		}
	}()
	go process.printShutdownStatus(localCtx)
	return localCtx
}

// Shutdown launches graceful shutdown process and waits
// for it to complete
func (process *TeleportProcess) Shutdown(ctx context.Context) {
	localCtx := process.StartShutdown(ctx)
	// wait until parent context closes
	<-localCtx.Done()
	process.log.Debug("Process completed.")
}

// Close broadcasts close signals and exits immediately
func (process *TeleportProcess) Close() error {
	process.BroadcastEvent(Event{Name: TeleportExitEvent})

	process.Config.Keygen.Close()

	var errors []error
	localAuth := process.getLocalAuth()
	if localAuth != nil {
		errors = append(errors, process.localAuth.Close())
	}

	if process.storage != nil {
		errors = append(errors, process.storage.Close())
	}

	return trace.NewAggregate(errors...)
}

func validateConfig(cfg *Config) error {
	if !cfg.Auth.Enabled && !cfg.SSH.Enabled && !cfg.Proxy.Enabled && !cfg.Kube.Enabled && !cfg.Apps.Enabled && !cfg.Databases.Enabled {
		return trace.BadParameter(
			"config: supply at least one of Auth, SSH, Proxy, App or Database roles")
	}

	if cfg.DataDir == "" {
		return trace.BadParameter("config: please supply data directory")
	}

	if cfg.Console == nil {
		cfg.Console = ioutil.Discard
	}

	if len(cfg.AuthServers) == 0 {
		return trace.BadParameter("auth_servers is empty")
	}
	for i := range cfg.Auth.Authorities {
		if err := services.ValidateCertAuthority(cfg.Auth.Authorities[i]); err != nil {
			return trace.Wrap(err)
		}
	}
	for _, tun := range cfg.ReverseTunnels {
		if err := services.ValidateReverseTunnel(tun); err != nil {
			return trace.Wrap(err)
		}
	}

	if cfg.PollingPeriod == 0 {
		cfg.PollingPeriod = defaults.LowResPollingPeriod
	}

	cfg.SSH.Namespace = types.ProcessNamespace(cfg.SSH.Namespace)

	return nil
}

// initSelfSignedHTTPSCert generates and self-signs a TLS key+cert pair for https connection
// to the proxy server.
func initSelfSignedHTTPSCert(cfg *Config) (err error) {
	cfg.Log.Warningf("No TLS Keys provided, using self-signed certificate.")

	keyPath := filepath.Join(cfg.DataDir, defaults.SelfSignedKeyPath)
	certPath := filepath.Join(cfg.DataDir, defaults.SelfSignedCertPath)

	cfg.Proxy.KeyPairs = append(cfg.Proxy.KeyPairs, KeyPairPath{
		PrivateKey:  keyPath,
		Certificate: certPath,
	})

	// return the existing pair if they have already been generated:
	_, err = tls.LoadX509KeyPair(certPath, keyPath)
	if err == nil {
		return nil
	}
	if !os.IsNotExist(err) {
		return trace.Wrap(err, "unrecognized error reading certs")
	}
	cfg.Log.Warningf("Generating self-signed key and cert to %v %v.", keyPath, certPath)

	creds, err := utils.GenerateSelfSignedCert([]string{cfg.Hostname, "localhost"})
	if err != nil {
		return trace.Wrap(err)
	}

	if err := ioutil.WriteFile(keyPath, creds.PrivateKey, 0600); err != nil {
		return trace.Wrap(err, "error writing key PEM")
	}
	if err := ioutil.WriteFile(certPath, creds.Cert, 0600); err != nil {
		return trace.Wrap(err, "error writing key PEM")
	}
	return nil
}

// initDebugApp starts a debug server that dumpers request headers.
func (process *TeleportProcess) initDebugApp() {
	var server *httptest.Server

	process.RegisterFunc("debug.app.service", func() error {
		server = httptest.NewServer(http.HandlerFunc(dumperHandler))
		process.BroadcastEvent(Event{Name: DebugAppReady, Payload: server})
		return nil
	})
	process.OnExit("debug.app.shutdown", func(payload interface{}) {
		if server != nil {
			server.Close()
		}
		process.log.Infof("Exited.")
	})
}

// singleProcessMode returns true when running all components needed within
// the same process. It's used for development and demo purposes.
func (process *TeleportProcess) singleProcessMode() (string, bool) {
	if !process.Config.Proxy.Enabled || !process.Config.Auth.Enabled {
		return "", false
	}
	if process.Config.Proxy.DisableReverseTunnel {
		return "", false
	}
	if len(process.Config.Proxy.TunnelPublicAddrs) == 0 {
		return net.JoinHostPort(string(teleport.PrincipalLocalhost), strconv.Itoa(defaults.SSHProxyTunnelListenPort)), true
	}
	return process.Config.Proxy.TunnelPublicAddrs[0].String(), true
}

// dumperHandler is an Application Access debugging application that will
// dump the headers of a request.
func dumperHandler(w http.ResponseWriter, r *http.Request) {
	requestDump, err := httputil.DumpRequest(r, true)
	if err != nil {
		http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
		return
	}

	randomBytes := make([]byte, 8)
	rand.Reader.Read(randomBytes)
	cookieValue := hex.EncodeToString(randomBytes)

	http.SetCookie(w, &http.Cookie{
		Name:     "dumper.session.cookie",
		Value:    cookieValue,
		SameSite: http.SameSiteLaxMode,
	})

	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
	fmt.Fprint(w, string(requestDump))
}

// getPublicAddr waits for a proxy to be registered with Teleport.
func getPublicAddr(authClient auth.AccessPoint, a App) (string, error) {
	ticker := time.NewTicker(500 * time.Millisecond)
	defer ticker.Stop()
	timeout := time.NewTimer(5 * time.Second)
	defer timeout.Stop()

	for {
		select {
		case <-ticker.C:
			publicAddr, err := findPublicAddr(authClient, a)
			if err == nil {
				return publicAddr, nil
			}
		case <-timeout.C:
			return "", trace.BadParameter("timed out waiting for proxy with public address")
		}
	}
}

// findPublicAddr tries to resolve the public address of the proxy of this cluster.
func findPublicAddr(authClient auth.AccessPoint, a App) (string, error) {
	// If the application has a public address already set, use it.
	if a.PublicAddr != "" {
		return a.PublicAddr, nil
	}

	// Fetch list of proxies, if first has public address set, use it.
	servers, err := authClient.GetProxies()
	if err != nil {
		return "", trace.Wrap(err)
	}
	if len(servers) == 0 {
		return "", trace.BadParameter("cluster has no proxied registered, at least one proxy must be registered for application access")
	}
	if servers[0].GetPublicAddr() != "" {
		addr, err := utils.ParseAddr(servers[0].GetPublicAddr())
		if err != nil {
			return "", trace.Wrap(err)
		}
		return fmt.Sprintf("%v.%v", a.Name, addr.Host()), nil
	}

	// Fall back to cluster name.
	cn, err := authClient.GetClusterName()
	if err != nil {
		return "", trace.Wrap(err)
	}
	return fmt.Sprintf("%v.%v", a.Name, cn.GetClusterName()), nil
}

// newHTTPFileSystem creates a new HTTP file system for the web handler.
// It uses external configuration to make the decision
func newHTTPFileSystem() (http.FileSystem, error) {
	if !isDebugMode() {
		fs, err := web.NewStaticFileSystem() //nolint:staticcheck
		if err != nil {                      //nolint:staticcheck
			return nil, trace.Wrap(err)
		}
		return fs, nil
	}
	// Use debug HTTP file system with default assets path
	fs, err := web.NewDebugFileSystem("")
	if err != nil {
		return nil, trace.Wrap(err)
	}
	return fs, nil
}

// isDebugMode determines if teleport is running in a "debug" mode.
// It looks at DEBUG environment variable
func isDebugMode() bool {
	v, _ := strconv.ParseBool(os.Getenv(teleport.DebugEnvVar))
	return v
}