Adding cache for recording_encryption resource

api/client/events.go

@@ -30,6 +30,7 @@ import (
 	notificationsv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/notifications/v1"
 	presencev1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/presence/v1"
 	provisioningv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/provisioning/v1"
+	recordingencryptionv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/recordingencryption/v1"
 	accessv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/scopes/access/v1"
 	userprovisioningpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/userprovisioning/v2"
 	usertasksv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/usertasks/v1"
@@ -157,6 +158,10 @@ func EventToGRPC(in types.Event) (*proto.Event, error) {
 		out.Resource = &proto.Event_WorkloadIdentityX509Revocation{
 			WorkloadIdentityX509Revocation: r.UnwrapT(),
 		}
+	case types.Resource153UnwrapperT[*recordingencryptionv1.RecordingEncryption]:
+		out.Resource = &proto.Event_RecordingEncryption{
+			RecordingEncryption: r.UnwrapT(),
+		}
 	case types.Resource153UnwrapperT[*healthcheckconfigv1.HealthCheckConfig]:
 		out.Resource = &proto.Event_HealthCheckConfig{
 			HealthCheckConfig: r.UnwrapT(),
@@ -660,6 +665,9 @@ func EventFromGRPC(in *proto.Event) (*types.Event, error) {
 	} else if r := in.GetRelayServer(); r != nil {
 		out.Resource = types.ProtoResource153ToLegacy(r)
 		return &out, nil
+	} else if r := in.GetRecordingEncryption(); r != nil {
+		out.Resource = types.ProtoResource153ToLegacy(r)
+		return &out, nil
 	} else {
 		return nil, trace.BadParameter("received unsupported resource %T", in.Resource)
 	}

api/proto/teleport/legacy/client/proto/event.proto

@@ -32,6 +32,7 @@ import "teleport/machineid/v1/federation.proto";
 import "teleport/notifications/v1/notifications.proto";
 import "teleport/presence/v1/relay_server.proto";
 import "teleport/provisioning/v1/provisioning.proto";
+import "teleport/recordingencryption/v1/recording_encryption.proto";
 import "teleport/scopes/access/v1/assignment.proto";
 import "teleport/scopes/access/v1/role.proto";
 import "teleport/secreports/v1/secreports.proto";
@@ -227,5 +228,7 @@ message Event {
     // ScopedRoleAssignment is an assignment of one or more scoped roles to a user.
     teleport.scopes.access.v1.ScopedRoleAssignment ScopedRoleAssignment = 81;
     teleport.presence.v1.RelayServer relay_server = 82;
+    // RecordingEncryption is a resource for controlling session recording encryption.
+    teleport.recordingencryption.v1.RecordingEncryption RecordingEncryption = 83;
   }
 }

lib/auth/accesspoint/accesspoint.go

@@ -112,6 +112,7 @@ type Config struct {
 	PluginStaticCredentials services.PluginStaticCredentials
 	GitServers              services.GitServers
 	HealthCheckConfig       services.HealthCheckConfigReader
+	RecordingEncryption     services.RecordingEncryption
 }
 
 func (c *Config) CheckAndSetDefaults() error {
@@ -213,6 +214,7 @@ func NewCache(cfg Config) (*cache.Cache, error) {
 		GitServers:              cfg.GitServers,
 		HealthCheckConfig:       cfg.HealthCheckConfig,
 		BotInstanceService:      cfg.BotInstance,
+		RecordingEncryption:     cfg.RecordingEncryption,
 	}
 
 	return cache.New(cfg.Setup(cacheCfg))

lib/auth/auth.go

@@ -256,6 +256,7 @@ func NewServer(cfg *InitConfig, opts ...ServerOption) (*Server, error) {
 
 		recordingEncryptionManager, err := recordingencryption.NewManager(recordingencryption.ManagerConfig{
 			Backend:       localRecordingEncryption,
+			Cache:         localRecordingEncryption,
 			ClusterConfig: cfg.ClusterConfiguration,
 			KeyStore:      cfg.KeyStore,
 			Logger:        cfg.Logger,

lib/auth/helpers.go

@@ -553,6 +553,7 @@ func InitTestAuthCache(p TestAuthCacheParams) error {
 		GitServers:              p.AuthServer.Services.GitServers,
 		HealthCheckConfig:       p.AuthServer.Services.HealthCheckConfig,
 		BotInstance:             p.AuthServer.Services.BotInstance,
+		RecordingEncryption:     p.AuthServer.Services.RecordingEncryptionManager,
 	})
 	if err != nil {
 		return trace.Wrap(err)

lib/auth/init.go

@@ -91,6 +91,7 @@ type VersionStorage interface {
 type RecordingEncryptionManager interface {
 	services.RecordingEncryption
 	recordingencryption.DecryptionKeyFinder
+	SetCache(cache recordingencryption.Cache)
 }
 
 // InitConfig is auth server init config

lib/auth/recordingencryption/manager.go

@@ -46,16 +46,22 @@ type KeyStore interface {
 	GetDecrypter(ctx context.Context, keyPair *types.EncryptionKeyPair) (crypto.Decrypter, error)
 }
 
+// A Cache fetches a cached [*recordingencryptionv1.RecordingEncryption].
+type Cache interface {
+	GetRecordingEncryption(context.Context) (*recordingencryptionv1.RecordingEncryption, error)
+}
+
 // ManagerConfig captures all of the dependencies required to instantiate a Manager.
 type ManagerConfig struct {
 	Backend       services.RecordingEncryption
 	ClusterConfig services.ClusterConfigurationInternal
 	KeyStore      KeyStore
+	Cache         Cache
 	Logger        *slog.Logger
 	LockConfig    backend.RunWhileLockedConfig
 }
 
-// NewManager returns a new Manager using the given ManagerConfig.
+// NewManager returns a new Manager using the given [ManagerConfig].
 func NewManager(cfg ManagerConfig) (*Manager, error) {
 	switch {
 	case cfg.Backend == nil:
@@ -64,6 +70,8 @@ func NewManager(cfg ManagerConfig) (*Manager, error) {
 		return nil, trace.BadParameter("cluster config is required")
 	case cfg.KeyStore == nil:
 		return nil, trace.BadParameter("key store is required")
+	case cfg.Cache == nil:
+		return nil, trace.BadParameter("cache is required")
 	}
 
 	if cfg.Logger == nil {
@@ -74,6 +82,7 @@ func NewManager(cfg ManagerConfig) (*Manager, error) {
 		RecordingEncryption:          cfg.Backend,
 		ClusterConfigurationInternal: cfg.ClusterConfig,
 
+		cache:      cfg.Cache,
 		keyStore:   cfg.KeyStore,
 		lockConfig: cfg.LockConfig,
 		logger:     cfg.Logger,
@@ -87,6 +96,7 @@ type Manager struct {
 	services.RecordingEncryption
 	services.ClusterConfigurationInternal
 
+	cache      Cache
 	logger     *slog.Logger
 	lockConfig backend.RunWhileLockedConfig
 	keyStore   KeyStore
@@ -164,6 +174,11 @@ func (m *Manager) UpsertSessionRecordingConfig(ctx context.Context, cfg types.Se
 	return sessionRecordingConfig, trace.Wrap(err)
 }
 
+// SetCache overwrites the configured Cache implementation. It should only be called if the `Manager` is not in use.
+func (m *Manager) SetCache(cache Cache) {
+	m.cache = cache
+}
+
 // ensureActiveRecordingEncryption returns the configured RecordingEncryption resource if it exists with active keys. If it does not,
 // then the resource will be created or updated with a new active keypair. The bool return value indicates whether or not
 // a new pair was provisioned.
@@ -367,7 +382,7 @@ func (m *Manager) searchActiveKeys(ctx context.Context, activeKeys []*recordinge
 
 // FindDecryptionKey returns the first accessible decryption key that matches one of the given public keys.
 func (m *Manager) FindDecryptionKey(ctx context.Context, publicKeys ...[]byte) (*types.EncryptionKeyPair, error) {
-	encryption, err := m.RecordingEncryption.GetRecordingEncryption(ctx)
+	encryption, err := m.cache.GetRecordingEncryption(ctx)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}

lib/auth/recordingencryption/manager_test.go

@@ -134,6 +134,7 @@ func newManagerConfig(t *testing.T, bk backend.Backend, keyType types.PrivateKey
 
 	return recordingencryption.ManagerConfig{
 		Backend:       recordingEncryptionService,
+		Cache:         recordingEncryptionService,
 		ClusterConfig: clusterConfigService,
 		KeyStore:      &fakeKeyStore{keyType: keyType},
 		Logger:        utils.NewSlogLoggerForTests(),

lib/cache/cache.go

@@ -211,6 +211,7 @@ func ForAuth(cfg Config) Config {
 		{Kind: types.KindHealthCheckConfig},
 		{Kind: types.KindRelayServer},
 		{Kind: types.KindBotInstance},
+		{Kind: types.KindRecordingEncryption},
 	}
 	cfg.QueueSize = defaults.AuthQueueSize
 	// We don't want to enable partial health for auth cache because auth uses an event stream
@@ -747,6 +748,8 @@ type Config struct {
 	HealthCheckConfig services.HealthCheckConfigReader
 	// BotInstanceService is the upstream service that we're caching
 	BotInstanceService services.BotInstance
+	// RecordingEncryption manages state surrounding session recording encryption
+	RecordingEncryption services.RecordingEncryption
 }
 
 // CheckAndSetDefaults checks parameters and sets default values