diff --git a/lib/events/complete.go b/lib/events/complete.go index 56480c3940e..0b363ea216f 100644 --- a/lib/events/complete.go +++ b/lib/events/complete.go @@ -55,6 +55,8 @@ type UploadCompleterConfig struct { // GracePeriod is the period after which an upload's session // tracker will be check to see if it's an abandoned upload. GracePeriod time.Duration + // ClusterName identifies the originating teleport cluster + ClusterName string } // CheckAndSetDefaults checks and sets default values @@ -65,6 +67,9 @@ func (cfg *UploadCompleterConfig) CheckAndSetDefaults() error { if cfg.SessionTracker == nil { return trace.BadParameter("missing parameter SessionTracker") } + if cfg.ClusterName == "" { + return trace.BadParameter("missing parameter ClusterName") + } if cfg.Component == "" { cfg.Component = teleport.ComponentAuth } @@ -219,11 +224,12 @@ func (u *UploadCompleter) checkUploads(ctx context.Context) error { }() session := &events.SessionUpload{ Metadata: events.Metadata{ - Type: SessionUploadEvent, - Code: SessionUploadCode, - Time: u.cfg.Clock.Now().UTC(), - ID: uuid.New().String(), - Index: SessionUploadIndex, + Type: SessionUploadEvent, + Code: SessionUploadCode, + Time: u.cfg.Clock.Now().UTC(), + ID: uuid.New().String(), + Index: SessionUploadIndex, + ClusterName: u.cfg.ClusterName, }, SessionMetadata: events.SessionMetadata{ SessionID: string(uploadData.SessionID), diff --git a/lib/events/complete_test.go b/lib/events/complete_test.go index bfac2c6ef1a..bf155673b7f 100644 --- a/lib/events/complete_test.go +++ b/lib/events/complete_test.go @@ -66,6 +66,7 @@ func TestUploadCompleterCompletesAbandonedUploads(t *testing.T) { AuditLog: log, SessionTracker: sessionTrackerService, Clock: clock, + ClusterName: "teleport-cluster", }) require.NoError(t, err) @@ -106,6 +107,7 @@ func TestUploadCompleterWithGracePeriod(t *testing.T) { SessionTracker: sts, Clock: clock, GracePeriod: 2 * time.Hour, + ClusterName: "teleport-cluster", }) require.NoError(t, err) t.Cleanup(uc.Close) @@ -158,6 +160,7 @@ func TestUploadCompleterEmitsSessionEnd(t *testing.T) { AuditLog: log, Clock: clock, SessionTracker: &mockSessionTrackerService{}, + ClusterName: "teleport-cluster", }) require.NoError(t, err) diff --git a/lib/service/db.go b/lib/service/db.go index fc5d2b4d307..b008e9be251 100644 --- a/lib/service/db.go +++ b/lib/service/db.go @@ -28,7 +28,6 @@ import ( "github.com/gravitational/teleport/lib/reversetunnel" "github.com/gravitational/teleport/lib/services" "github.com/gravitational/teleport/lib/srv/db" - "github.com/gravitational/teleport/lib/utils" ) func (process *TeleportProcess) shouldInitDatabases() bool { @@ -74,6 +73,8 @@ func (process *TeleportProcess) initDatabaseService() (retErr error) { } } + clusterName := conn.ServerIdentity.ClusterName + // Start uploader that will scan a path on disk and upload completed // sessions to the auth server. uploaderCfg := filesessions.UploaderConfig{ @@ -82,6 +83,7 @@ func (process *TeleportProcess) initDatabaseService() (retErr error) { } completerCfg := events.UploadCompleterConfig{ SessionTracker: conn.Client, + ClusterName: clusterName, } err = process.initUploaderService(uploaderCfg, completerCfg) if err != nil { @@ -147,8 +149,6 @@ func (process *TeleportProcess) initDatabaseService() (retErr error) { databases = append(databases, db) } - clusterName := conn.ServerIdentity.Cert.Extensions[utils.CertExtensionAuthority] - lockWatcher, err := services.NewLockWatcher(process.ExitContext(), services.LockWatcherConfig{ ResourceWatcherConfig: services.ResourceWatcherConfig{ Component: teleport.ComponentDatabase, diff --git a/lib/service/kubernetes.go b/lib/service/kubernetes.go index 17735fc82e4..c30cd19fe51 100644 --- a/lib/service/kubernetes.go +++ b/lib/service/kubernetes.go @@ -71,6 +71,8 @@ func (process *TeleportProcess) initKubernetesService(log *logrus.Entry, conn *C return trace.Wrap(err) } + teleportClusterName := conn.ServerIdentity.ClusterName + // Start uploader that will scan a path on disk and upload completed // sessions to the Auth Server. uploaderCfg := filesessions.UploaderConfig{ @@ -79,6 +81,7 @@ func (process *TeleportProcess) initKubernetesService(log *logrus.Entry, conn *C } completerCfg := events.UploadCompleterConfig{ SessionTracker: conn.Client, + ClusterName: teleportClusterName, } if err := process.initUploaderService(uploaderCfg, completerCfg); err != nil { return trace.Wrap(err) @@ -138,7 +141,7 @@ func (process *TeleportProcess) initKubernetesService(log *logrus.Entry, conn *C Client: conn.Client, AccessPoint: accessPoint, HostSigner: conn.ServerIdentity.KeySigner, - Cluster: conn.ServerIdentity.Cert.Extensions[utils.CertExtensionAuthority], + Cluster: teleportClusterName, Server: shtl, FIPS: process.Config.FIPS, ConnectedProxyGetter: proxyGetter, @@ -175,8 +178,6 @@ func (process *TeleportProcess) initKubernetesService(log *logrus.Entry, conn *C }() } - teleportClusterName := conn.ServerIdentity.Cert.Extensions[utils.CertExtensionAuthority] - lockWatcher, err := services.NewLockWatcher(process.ExitContext(), services.LockWatcherConfig{ ResourceWatcherConfig: services.ResourceWatcherConfig{ Component: teleport.ComponentKube, diff --git a/lib/service/service.go b/lib/service/service.go index aac3d2e24d1..90f0f997ae2 100644 --- a/lib/service/service.go +++ b/lib/service/service.go @@ -1489,11 +1489,11 @@ func (process *TeleportProcess) initAuthService() error { emitter = localLog } } - + clusterName := cfg.Auth.ClusterName.GetClusterName() checkingEmitter, err := events.NewCheckingEmitter(events.CheckingEmitterConfig{ Inner: events.NewMultiEmitter(events.NewLoggingEmitter(), emitter), Clock: process.Clock, - ClusterName: cfg.Auth.ClusterName.GetClusterName(), + ClusterName: clusterName, }) if err != nil { return trace.Wrap(err) @@ -1502,7 +1502,7 @@ func (process *TeleportProcess) initAuthService() error { checkingStreamer, err := events.NewCheckingStreamer(events.CheckingStreamerConfig{ Inner: streamer, Clock: process.Clock, - ClusterName: cfg.Auth.ClusterName.GetClusterName(), + ClusterName: clusterName, }) if err != nil { return trace.Wrap(err) @@ -1605,6 +1605,7 @@ func (process *TeleportProcess) initAuthService() error { Component: teleport.ComponentAuth, AuditLog: process.auditLog, SessionTracker: authServer.Services, + ClusterName: clusterName, // DELETE IN 11.0.0 // Provide a grace period so that Auth does not prematurely upload // sessions which don't have a session tracker (v9.2 and earlier) @@ -1627,7 +1628,7 @@ func (process *TeleportProcess) initAuthService() error { if err != nil { return trace.Wrap(err) } - authorizer, err := auth.NewAuthorizer(cfg.Auth.ClusterName.GetClusterName(), authServer, lockWatcher) + authorizer, err := auth.NewAuthorizer(clusterName, authServer, lockWatcher) if err != nil { return trace.Wrap(err) } @@ -2309,6 +2310,7 @@ func (process *TeleportProcess) initSSH() error { completerCfg := events.UploadCompleterConfig{ SessionTracker: conn.Client, GracePeriod: defaults.UploadGracePeriod, + ClusterName: conn.ServerIdentity.ClusterName, } if err := process.initUploaderService(uploaderCfg, completerCfg); err != nil { return trace.Wrap(err) @@ -3244,7 +3246,7 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error { trace.Component: teleport.Component(teleport.ComponentReverseTunnelServer, process.id), }) - clusterName := conn.ServerIdentity.Cert.Extensions[utils.CertExtensionAuthority] + clusterName := conn.ServerIdentity.ClusterName // asyncEmitter makes sure that sessions do not block // in case if connections are slow @@ -3859,6 +3861,7 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error { } completerCfg := events.UploadCompleterConfig{ SessionTracker: conn.Client, + ClusterName: clusterName, } if err := process.initUploaderService(uploaderCfg, completerCfg); err != nil { return trace.Wrap(err) @@ -4153,6 +4156,8 @@ func (process *TeleportProcess) initApps() { log.Debugf("Application service dependencies have started, continuing.") } + clusterName := conn.ServerIdentity.ClusterName + // Start uploader that will scan a path on disk and upload completed // sessions to the Auth Server. uploaderCfg := filesessions.UploaderConfig{ @@ -4161,6 +4166,7 @@ func (process *TeleportProcess) initApps() { } completerCfg := events.UploadCompleterConfig{ SessionTracker: conn.Client, + ClusterName: clusterName, } if err := process.initUploaderService(uploaderCfg, completerCfg); err != nil { return trace.Wrap(err) @@ -4234,8 +4240,6 @@ func (process *TeleportProcess) initApps() { applications = append(applications, a) } - clusterName := conn.ServerIdentity.Cert.Extensions[utils.CertExtensionAuthority] - lockWatcher, err := services.NewLockWatcher(process.ExitContext(), services.LockWatcherConfig{ ResourceWatcherConfig: services.ResourceWatcherConfig{ Component: teleport.ComponentApp,