opensource/teleport
3
0
Fork 0
mirror of https://github.com/gravitational/teleport.git synced 2026-10-29 17:58:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix missing cluster name on session.upload via Upload Completer (#15120)

Browse Source
This commit is contained in:
Tobiasz Heller
2022-08-05 13:56:38 +02:00
committed by GitHub
parent 61c39a9e2b
commit 57d572dcc5
5 changed files with 32 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/events/complete.go
+11 -5
View File
@@ -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),
lib/events/complete_test.go
+3
View File
@@ -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)
lib/service/db.go
+3 -3
View File
@@ -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,
lib/service/kubernetes.go
+4 -3
View File
@@ -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,
lib/service/service.go
+11 -7
View File
@@ -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,