[v12] Track user connections across clusters

integration/appaccess/pack.go

@@ -44,7 +44,7 @@ import (
 	"github.com/gravitational/teleport/lib/auth/native"
 	"github.com/gravitational/teleport/lib/client"
 	"github.com/gravitational/teleport/lib/httplib/csrf"
-	"github.com/gravitational/teleport/lib/reversetunnel"
+	"github.com/gravitational/teleport/lib/reversetunnelclient"
 	"github.com/gravitational/teleport/lib/service"
 	"github.com/gravitational/teleport/lib/services"
 	"github.com/gravitational/teleport/lib/srv/alpnproxy"
@@ -750,7 +750,7 @@ func (p *Pack) startRootAppServers(t *testing.T, count int, opts AppTestOptions)
 	return servers
 }
 
-func waitForAppServer(t *testing.T, tunnel reversetunnel.Server, name string, hostUUID string, apps []service.App) {
+func waitForAppServer(t *testing.T, tunnel reversetunnelclient.Server, name string, hostUUID string, apps []service.App) {
 	// Make sure that the app server is ready to accept connections.
 	// The remote site cache needs to be filled with new registered application services.
 	waitForAppRegInRemoteSiteCache(t, tunnel, name, apps, hostUUID)
@@ -888,7 +888,7 @@ func (p *Pack) startLeafAppServers(t *testing.T, count int, opts AppTestOptions)
 	return servers
 }
 
-func waitForAppRegInRemoteSiteCache(t *testing.T, tunnel reversetunnel.Server, clusterName string, cfgApps []service.App, hostUUID string) {
+func waitForAppRegInRemoteSiteCache(t *testing.T, tunnel reversetunnelclient.Server, clusterName string, cfgApps []service.App, hostUUID string) {
 	require.Eventually(t, func() bool {
 		site, err := tunnel.GetSite(clusterName)
 		require.NoError(t, err)

integration/helpers/instance.go

@@ -56,6 +56,7 @@ import (
 	"github.com/gravitational/teleport/lib/httplib/csrf"
 	"github.com/gravitational/teleport/lib/observability/tracing"
 	"github.com/gravitational/teleport/lib/reversetunnel"
+	"github.com/gravitational/teleport/lib/reversetunnelclient"
 	"github.com/gravitational/teleport/lib/service"
 	"github.com/gravitational/teleport/lib/services"
 	"github.com/gravitational/teleport/lib/session"
@@ -256,7 +257,7 @@ type TeleInstance struct {
 	// Internal stuff...
 	Process              *service.TeleportProcess
 	Config               *service.Config
-	Tunnel               reversetunnel.Server
+	Tunnel               reversetunnelclient.Server
 	RemoteClusterWatcher *reversetunnel.RemoteClusterTunnelManager
 
 	// Nodes is a list of additional nodes
@@ -1029,7 +1030,7 @@ type ProxyConfig struct {
 }
 
 // StartProxy starts another Proxy Server and connects it to the cluster.
-func (i *TeleInstance) StartProxy(cfg ProxyConfig) (reversetunnel.Server, *service.TeleportProcess, error) {
+func (i *TeleInstance) StartProxy(cfg ProxyConfig) (reversetunnelclient.Server, *service.TeleportProcess, error) {
 	dataDir, err := os.MkdirTemp("", "cluster-"+i.Secrets.SiteName+"-"+cfg.Name)
 	if err != nil {
 		return nil, nil, trace.Wrap(err)
@@ -1098,12 +1099,12 @@ func (i *TeleInstance) StartProxy(cfg ProxyConfig) (reversetunnel.Server, *servi
 	log.Debugf("Teleport proxy (in instance %v) started: %v/%v events received.",
 		i.Secrets.SiteName, len(expectedEvents), len(receivedEvents))
 
-	// Extract and set reversetunnel.Server and reversetunnel.AgentPool upon
+	// Extract and set reversetunnelclient.Server and reversetunnel.AgentPool upon
 	// receipt of a ProxyReverseTunnelReady event
 	for _, re := range receivedEvents {
 		switch re.Name {
 		case service.ProxyReverseTunnelReady:
-			ts, ok := re.Payload.(reversetunnel.Server)
+			ts, ok := re.Payload.(reversetunnelclient.Server)
 			if ok {
 				return ts, process, nil
 			}
@@ -1199,12 +1200,12 @@ func (i *TeleInstance) Start() error {
 		return trace.Wrap(err)
 	}
 
-	// Extract and set reversetunnel.Server and reversetunnel.AgentPool upon
+	// Extract and set reversetunnelclient.Server and reversetunnel.AgentPool upon
 	// receipt of a ProxyReverseTunnelReady and ProxyAgentPoolReady respectively.
 	for _, re := range receivedEvents {
 		switch re.Name {
 		case service.ProxyReverseTunnelReady:
-			ts, ok := re.Payload.(reversetunnel.Server)
+			ts, ok := re.Payload.(reversetunnelclient.Server)
 			if ok {
 				i.Tunnel = ts
 			}

integration/helpers/trustedclusters.go

@@ -28,7 +28,7 @@ import (
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/api/utils/retryutils"
 	"github.com/gravitational/teleport/lib/auth"
-	"github.com/gravitational/teleport/lib/reversetunnel"
+	"github.com/gravitational/teleport/lib/reversetunnelclient"
 )
 
 // WaitForTunnelConnections waits for remote tunnels connections
@@ -76,7 +76,7 @@ func TryCreateTrustedCluster(t *testing.T, authServer *auth.Server, trustedClust
 	require.FailNow(t, "Timeout creating trusted cluster")
 }
 
-func WaitForClusters(tun reversetunnel.Server, expected int) func() bool {
+func WaitForClusters(tun reversetunnelclient.Server, expected int) func() bool {
 	return func() bool {
 		clusters, err := tun.GetSites()
 		if err != nil {
@@ -129,7 +129,7 @@ func WaitForNodeCount(ctx context.Context, t *TeleInstance, clusterName string,
 }
 
 // WaitForActiveTunnelConnections waits for remote cluster to report a minimum number of active connections
-func WaitForActiveTunnelConnections(t *testing.T, tunnel reversetunnel.Server, clusterName string, expectedCount int) {
+func WaitForActiveTunnelConnections(t *testing.T, tunnel reversetunnelclient.Server, clusterName string, expectedCount int) {
 	require.Eventually(t, func() bool {
 		cluster, err := tunnel.GetSite(clusterName)
 		if err != nil {

integration/integration_test.go

@@ -77,7 +77,7 @@ import (
 	"github.com/gravitational/teleport/lib/events/filesessions"
 	"github.com/gravitational/teleport/lib/modules"
 	"github.com/gravitational/teleport/lib/pam"
-	"github.com/gravitational/teleport/lib/reversetunnel"
+	"github.com/gravitational/teleport/lib/reversetunnelclient"
 	"github.com/gravitational/teleport/lib/service"
 	"github.com/gravitational/teleport/lib/services"
 	"github.com/gravitational/teleport/lib/session"
@@ -3520,7 +3520,7 @@ func testDiscoveryRecovers(t *testing.T, suite *integrationTestSuite) {
 	var reverseTunnelAddr string
 
 	// Helper function for adding a new proxy to "main".
-	addNewMainProxy := func(name string) (reversetunnel.Server, helpers.ProxyConfig) {
+	addNewMainProxy := func(name string) (reversetunnelclient.Server, helpers.ProxyConfig) {
 		t.Logf("adding main proxy %q...", name)
 		newConfig := helpers.ProxyConfig{
 			Name:              name,

integration/proxy/proxy_helpers.go

@@ -44,7 +44,7 @@ import (
 	"github.com/gravitational/teleport/api/utils/retryutils"
 	"github.com/gravitational/teleport/integration/helpers"
 	"github.com/gravitational/teleport/lib/defaults"
-	"github.com/gravitational/teleport/lib/reversetunnel"
+	"github.com/gravitational/teleport/lib/reversetunnelclient"
 	"github.com/gravitational/teleport/lib/service"
 	"github.com/gravitational/teleport/lib/services"
 	"github.com/gravitational/teleport/lib/srv/alpnproxy"
@@ -587,7 +587,7 @@ func mustStartMockALBProxy(t *testing.T, proxyAddr string) *mockAWSALBProxy {
 }
 
 // waitForActivePeerProxyConnections waits for remote cluster to report a minimum number of active proxy peer connections
-func waitForActivePeerProxyConnections(t *testing.T, tunnel reversetunnel.Server, expectedCount int) { //nolint:unused // Only used by skipped test TestProxyTunnelStrategyProxyPeering
+func waitForActivePeerProxyConnections(t *testing.T, tunnel reversetunnelclient.Server, expectedCount int) { //nolint:unused // Only used by skipped test TestProxyTunnelStrategyProxyPeering
 	require.Eventually(t, func() bool {
 		return tunnel.GetProxyPeerClient().GetConnectionsCount() >= expectedCount
 	},

lib/auth/authclient/authclient.go

@@ -31,7 +31,7 @@ import (
 	apiclient "github.com/gravitational/teleport/api/client"
 	"github.com/gravitational/teleport/api/client/webclient"
 	"github.com/gravitational/teleport/lib/auth"
-	"github.com/gravitational/teleport/lib/reversetunnel"
+	"github.com/gravitational/teleport/lib/reversetunnelclient"
 	"github.com/gravitational/teleport/lib/utils"
 )
 
@@ -110,21 +110,21 @@ func connectViaProxyTunnel(ctx context.Context, cfg *Config) (auth.ClientI, erro
 	//
 	// TODO(nic): this logic should be implemented once and reused in IoT
 	// nodes.
-	resolver := reversetunnel.WebClientResolver(&webclient.Config{
+	resolver := reversetunnelclient.WebClientResolver(&webclient.Config{
 		Context:   ctx,
 		ProxyAddr: cfg.AuthServers[0].String(),
 		Insecure:  cfg.TLS.InsecureSkipVerify,
 		Timeout:   cfg.DialTimeout,
 	})
 
-	resolver, err := reversetunnel.CachingResolver(ctx, resolver, nil /* clock */)
+	resolver, err := reversetunnelclient.CachingResolver(ctx, resolver, nil /* clock */)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
 
-	// reversetunnel.TunnelAuthDialer will take care of creating a net.Conn
+	// reversetunnelclient.TunnelAuthDialer will take care of creating a net.Conn
 	// within an SSH tunnel.
-	dialer, err := reversetunnel.NewTunnelAuthDialer(reversetunnel.TunnelAuthDialerConfig{
+	dialer, err := reversetunnelclient.NewTunnelAuthDialer(reversetunnelclient.TunnelAuthDialerConfig{
 		Resolver:              resolver,
 		ClientConfig:          cfg.SSH,
 		Log:                   cfg.Log,

lib/kube/proxy/forwarder.go

@@ -79,7 +79,7 @@ import (
 	"github.com/gravitational/teleport/lib/kube/proxy/streamproto"
 	kubeutils "github.com/gravitational/teleport/lib/kube/utils"
 	"github.com/gravitational/teleport/lib/multiplexer"
-	"github.com/gravitational/teleport/lib/reversetunnel"
+	"github.com/gravitational/teleport/lib/reversetunnelclient"
 	"github.com/gravitational/teleport/lib/services"
 	"github.com/gravitational/teleport/lib/srv"
 	"github.com/gravitational/teleport/lib/sshca"
@@ -107,7 +107,7 @@ const (
 // ForwarderConfig specifies configuration for proxy forwarder
 type ForwarderConfig struct {
 	// ReverseTunnelSrv is the teleport reverse tunnel server
-	ReverseTunnelSrv reversetunnel.Server
+	ReverseTunnelSrv reversetunnelclient.Server
 	// ClusterName is a local cluster name
 	ClusterName string
 	// Keygen points to a key generator implementation
@@ -718,7 +718,7 @@ func (f *Forwarder) setupContext(authCtx authz.Context, req *http.Request, isRem
 		}
 
 		dialFn = func(ctx context.Context, network string, endpoint kubeClusterEndpoint) (net.Conn, error) {
-			return targetCluster.DialTCP(reversetunnel.DialParams{
+			return targetCluster.DialTCP(reversetunnelclient.DialParams{
 				From:                  &utils.NetAddr{AddrNetwork: "tcp", Addr: req.RemoteAddr},
 				To:                    &utils.NetAddr{AddrNetwork: "tcp", Addr: endpoint.addr},
 				ConnType:              types.KubeTunnel,
@@ -730,7 +730,7 @@ func (f *Forwarder) setupContext(authCtx authz.Context, req *http.Request, isRem
 		isRemoteClosed = targetCluster.IsClosed
 	} else if f.cfg.ReverseTunnelSrv != nil {
 		// Not a remote cluster and we have a reverse tunnel server.
-		// Use the local reversetunnel.Site which knows how to dial by serverID
+		// Use the local reversetunnelclient.Site which knows how to dial by serverID
 		// (for "kubernetes_service" connected over a tunnel) and falls back to
 		// direct dial if needed.
 		localCluster, err := f.cfg.ReverseTunnelSrv.GetSite(f.cfg.ClusterName)
@@ -743,7 +743,7 @@ func (f *Forwarder) setupContext(authCtx authz.Context, req *http.Request, isRem
 			if forwarderType != ProxyService && endpoint.serverID == "" {
 				clientDst = nil
 			}
-			return localCluster.DialTCP(reversetunnel.DialParams{
+			return localCluster.DialTCP(reversetunnelclient.DialParams{
 				From:                  &utils.NetAddr{AddrNetwork: "tcp", Addr: req.RemoteAddr},
 				To:                    &utils.NetAddr{AddrNetwork: "tcp", Addr: endpoint.addr},
 				ConnType:              types.KubeTunnel,
@@ -2120,7 +2120,7 @@ func (f *Forwarder) newClusterSessionRemoteCluster(ctx authContext) (*clusterSes
 		// and the targetKubernetes cluster endpoint is determined from the identity
 		// encoded in the TLS certificate. We're setting the dial endpoint to a hardcoded
 		// `kube.teleport.cluster.local` value to indicate this is a Kubernetes proxy request
-		kubeClusterEndpoints: []kubeClusterEndpoint{{addr: reversetunnel.LocalKubernetes}},
+		kubeClusterEndpoints: []kubeClusterEndpoint{{addr: reversetunnelclient.LocalKubernetes}},
 		tlsConfig:            tlsConfig.Clone(),
 	}, nil
 }

lib/kube/proxy/forwarder_test.go

@@ -53,7 +53,7 @@ import (
 	"github.com/gravitational/teleport/lib/backend/memory"
 	"github.com/gravitational/teleport/lib/defaults"
 	"github.com/gravitational/teleport/lib/fixtures"
-	"github.com/gravitational/teleport/lib/reversetunnel"
+	"github.com/gravitational/teleport/lib/reversetunnelclient"
 	"github.com/gravitational/teleport/lib/services"
 	"github.com/gravitational/teleport/lib/services/local"
 	"github.com/gravitational/teleport/lib/tlsca"
@@ -148,7 +148,7 @@ func TestAuthenticate(t *testing.T) {
 	require.NoError(t, err)
 
 	tun := mockRevTunnel{
-		sites: map[string]reversetunnel.RemoteSite{
+		sites: map[string]reversetunnelclient.RemoteSite{
 			"remote": mockRemoteSite{name: "remote"},
 			"local":  mockRemoteSite{name: "local"},
 		},
@@ -185,7 +185,7 @@ func TestAuthenticate(t *testing.T) {
 		routeToCluster    string
 		kubernetesCluster string
 		haveKubeCreds     bool
-		tunnel            reversetunnel.Server
+		tunnel            reversetunnelclient.Server
 		kubeServers       []types.KubeServer
 		activeRequests    []string
 
@@ -1034,7 +1034,7 @@ func TestNewClusterSessionRemote(t *testing.T) {
 	// Succeed on remote cluster session
 	sess, err := f.newClusterSession(authCtx)
 	require.NoError(t, err)
-	require.Equal(t, []kubeClusterEndpoint{{addr: reversetunnel.LocalKubernetes}}, sess.kubeClusterEndpoints)
+	require.Equal(t, []kubeClusterEndpoint{{addr: reversetunnelclient.LocalKubernetes}}, sess.kubeClusterEndpoints)
 
 	// Make sure newClusterSession obtained a new client cert instead of using f.creds.
 	require.Equal(t, f.cfg.AuthClient.(*mockCSRClient).lastCert.Raw, sess.tlsConfig.Certificates[0].Certificate[0])
@@ -1085,7 +1085,7 @@ func TestNewClusterSessionDirect(t *testing.T) {
 
 	// multiple kube services for kube cluster
 	publicKubeService, publicEndpoint := newKubeServer("public", "k8s.example.com", "kube-cluster")
-	tunnelKubeService, tunnelEndpoint := newKubeServer("tunnel", reversetunnel.LocalKubernetes, "kube-cluster")
+	tunnelKubeService, tunnelEndpoint := newKubeServer("tunnel", reversetunnelclient.LocalKubernetes, "kube-cluster")
 
 	f.cfg.CachingAuthClient = mockAccessPoint{
 		kubeServers: []types.KubeServer{publicKubeService, otherKubeService, tunnelKubeService, otherKubeService},
@@ -1300,11 +1300,11 @@ func (c *mockCSRClient) ProcessKubeCSR(csr auth.KubeCSR) (*auth.KubeCSRResponse,
 	}, nil
 }
 
-// mockRemoteSite is a reversetunnel.RemoteSite implementation with hardcoded
+// mockRemoteSite is a reversetunnelclient.RemoteSite implementation with hardcoded
 // name, because there's no easy way to construct a real
-// reversetunnel.RemoteSite.
+// reversetunnelclient.RemoteSite.
 type mockRemoteSite struct {
-	reversetunnel.RemoteSite
+	reversetunnelclient.RemoteSite
 	name string
 }
 
@@ -1349,21 +1349,21 @@ func (ap mockAccessPoint) GetCertAuthority(ctx context.Context, id types.CertAut
 }
 
 type mockRevTunnel struct {
-	reversetunnel.Server
+	reversetunnelclient.Server
 
-	sites map[string]reversetunnel.RemoteSite
+	sites map[string]reversetunnelclient.RemoteSite
 }
 
-func (t mockRevTunnel) GetSite(name string) (reversetunnel.RemoteSite, error) {
+func (t mockRevTunnel) GetSite(name string) (reversetunnelclient.RemoteSite, error) {
 	s, ok := t.sites[name]
 	if !ok {
 		return nil, trace.NotFound("remote site %q not found", name)
 	}
 	return s, nil
 }
 
-func (t mockRevTunnel) GetSites() ([]reversetunnel.RemoteSite, error) {
-	var sites []reversetunnel.RemoteSite
+func (t mockRevTunnel) GetSites() ([]reversetunnelclient.RemoteSite, error) {
+	var sites []reversetunnelclient.RemoteSite
 	for _, s := range t.sites {
 		sites = append(sites, s)
 	}

lib/proxy/clusterdial/dial.go

@@ -20,7 +20,7 @@ import (
 	"github.com/gravitational/trace"
 
 	"github.com/gravitational/teleport/lib/proxy/peer"
-	"github.com/gravitational/teleport/lib/reversetunnel"
+	"github.com/gravitational/teleport/lib/reversetunnelclient"
 )
 
 // ClusterDialerFunc is a function that implements a peer.ClusterDialer.
@@ -32,14 +32,14 @@ func (f ClusterDialerFunc) Dial(clusterName string, request peer.DialParams) (ne
 }
 
 // NewClusterDialer implements proxy.ClusterDialer for a reverse tunnel server.
-func NewClusterDialer(server reversetunnel.Server) ClusterDialerFunc {
+func NewClusterDialer(server reversetunnelclient.Server) ClusterDialerFunc {
 	return func(clusterName string, request peer.DialParams) (net.Conn, error) {
 		site, err := server.GetSite(clusterName)
 		if err != nil {
 			return nil, trace.Wrap(err)
 		}
 
-		dialParams := reversetunnel.DialParams{
+		dialParams := reversetunnelclient.DialParams{
 			ServerID:      request.ServerID,
 			ConnType:      request.ConnType,
 			From:          request.From,