diff --git a/api/gen/proto/go/teleport/recordingencryption/v1/recording_encryption.pb.go b/api/gen/proto/go/teleport/recordingencryption/v1/recording_encryption.pb.go
index c6118aa023d75..8dc36719a1806 100644
--- a/api/gen/proto/go/teleport/recordingencryption/v1/recording_encryption.pb.go
+++ b/api/gen/proto/go/teleport/recordingencryption/v1/recording_encryption.pb.go
@@ -37,36 +37,29 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
-// WrappedKey wraps the private key of a recording encryption key pair using a
-// separate asymmetric keypair.
-type WrappedKey struct {
+// A key pair used with age to wrap and unwrap file keys for session recording encryption.
+type KeyPair struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
-	// RecordingEncryptionPair is the asymmetric keypair used with age to encrypt
-	// and decrypt filekeys. The private key is encrypted using the KeyEncryptionPair's
-	// public key and has to be decrypted before recording decryption operations can
-	// be fulfilled.
-	RecordingEncryptionPair *types.EncryptionKeyPair `protobuf:"bytes,1,opt,name=recording_encryption_pair,json=recordingEncryptionPair,proto3" json:"recording_encryption_pair,omitempty"`
-	// KeyEncryptionPair is the asymmetric keypair used to wrap (encrypt) the
-	// RecordingEncryptionPair's private key.
-	KeyEncryptionPair *types.EncryptionKeyPair `protobuf:"bytes,2,opt,name=key_encryption_pair,json=keyEncryptionPair,proto3" json:"key_encryption_pair,omitempty"`
-	unknownFields     protoimpl.UnknownFields
-	sizeCache         protoimpl.SizeCache
-}
-
-func (x *WrappedKey) Reset() {
-	*x = WrappedKey{}
+	// A key pair used with age to wrap and unwrap file keys for session recording encryption.
+	KeyPair       *types.EncryptionKeyPair `protobuf:"bytes,1,opt,name=key_pair,json=keyPair,proto3" json:"key_pair,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *KeyPair) Reset() {
+	*x = KeyPair{}
 	mi := &file_teleport_recordingencryption_v1_recording_encryption_proto_msgTypes[0]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
 
-func (x *WrappedKey) String() string {
+func (x *KeyPair) String() string {
 	return protoimpl.X.MessageStringOf(x)
 }
 
-func (*WrappedKey) ProtoMessage() {}
+func (*KeyPair) ProtoMessage() {}
 
-func (x *WrappedKey) ProtoReflect() protoreflect.Message {
+func (x *KeyPair) ProtoReflect() protoreflect.Message {
 	mi := &file_teleport_recordingencryption_v1_recording_encryption_proto_msgTypes[0]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -78,21 +71,14 @@ func (x *WrappedKey) ProtoReflect() protoreflect.Message {
 	return mi.MessageOf(x)
 }
 
-// Deprecated: Use WrappedKey.ProtoReflect.Descriptor instead.
-func (*WrappedKey) Descriptor() ([]byte, []int) {
+// Deprecated: Use KeyPair.ProtoReflect.Descriptor instead.
+func (*KeyPair) Descriptor() ([]byte, []int) {
 	return file_teleport_recordingencryption_v1_recording_encryption_proto_rawDescGZIP(), []int{0}
 }
 
-func (x *WrappedKey) GetRecordingEncryptionPair() *types.EncryptionKeyPair {
-	if x != nil {
-		return x.RecordingEncryptionPair
-	}
-	return nil
-}
-
-func (x *WrappedKey) GetKeyEncryptionPair() *types.EncryptionKeyPair {
+func (x *KeyPair) GetKeyPair() *types.EncryptionKeyPair {
 	if x != nil {
-		return x.KeyEncryptionPair
+		return x.KeyPair
 	}
 	return nil
 }
@@ -100,13 +86,13 @@ func (x *WrappedKey) GetKeyEncryptionPair() *types.EncryptionKeyPair {
 // RecordingEncryptionSpec contains the active key set for encrypted session recording.
 type RecordingEncryptionSpec struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
-	// AciveKeys is a list of active, wrapped X25519 keypairs. The unique set of RecordingEncryptionPair
-	// public keys are used as recipients during age encryption of session recordings. This
-	// allows any active private key to be used during decryption which guards against recordings
-	// being inaccessible to auth servers waiting for their key to rotate.
-	ActiveKeys    []*WrappedKey `protobuf:"bytes,1,rep,name=active_keys,json=activeKeys,proto3" json:"active_keys,omitempty"`
-	unknownFields protoimpl.UnknownFields
-	sizeCache     protoimpl.SizeCache
+	// A list of active key pairs used for session recording encryption. The unique set of
+	// active public keys are used as recipients during age encryption. This allows any
+	// active private key to be used during decryption which guards against recordings being
+	// inaccessible to auth servers waiting for key rotation.
+	ActiveKeyPairs []*KeyPair `protobuf:"bytes,2,rep,name=active_key_pairs,json=activeKeyPairs,proto3" json:"active_key_pairs,omitempty"`
+	unknownFields  protoimpl.UnknownFields
+	sizeCache      protoimpl.SizeCache
 }
 
 func (x *RecordingEncryptionSpec) Reset() {
@@ -139,9 +125,9 @@ func (*RecordingEncryptionSpec) Descriptor() ([]byte, []int) {
 	return file_teleport_recordingencryption_v1_recording_encryption_proto_rawDescGZIP(), []int{1}
 }
 
-func (x *RecordingEncryptionSpec) GetActiveKeys() []*WrappedKey {
+func (x *RecordingEncryptionSpec) GetActiveKeyPairs() []*KeyPair {
 	if x != nil {
-		return x.ActiveKeys
+		return x.ActiveKeyPairs
 	}
 	return nil
 }
@@ -272,14 +258,11 @@ var File_teleport_recordingencryption_v1_recording_encryption_proto protoreflect
 
 const file_teleport_recordingencryption_v1_recording_encryption_proto_rawDesc = "" +
 	"\n" +
-	":teleport/recordingencryption/v1/recording_encryption.proto\x12\x1fteleport.recordingencryption.v1\x1a!teleport/header/v1/metadata.proto\x1a!teleport/legacy/types/types.proto\"\xac\x01\n" +
-	"\n" +
-	"WrappedKey\x12T\n" +
-	"\x19recording_encryption_pair\x18\x01 \x01(\v2\x18.types.EncryptionKeyPairR\x17recordingEncryptionPair\x12H\n" +
-	"\x13key_encryption_pair\x18\x02 \x01(\v2\x18.types.EncryptionKeyPairR\x11keyEncryptionPair\"g\n" +
-	"\x17RecordingEncryptionSpec\x12L\n" +
-	"\vactive_keys\x18\x01 \x03(\v2+.teleport.recordingencryption.v1.WrappedKeyR\n" +
-	"activeKeys\"\x1b\n" +
+	":teleport/recordingencryption/v1/recording_encryption.proto\x12\x1fteleport.recordingencryption.v1\x1a!teleport/header/v1/metadata.proto\x1a!teleport/legacy/types/types.proto\">\n" +
+	"\aKeyPair\x123\n" +
+	"\bkey_pair\x18\x01 \x01(\v2\x18.types.EncryptionKeyPairR\akeyPair\"\x80\x01\n" +
+	"\x17RecordingEncryptionSpec\x12R\n" +
+	"\x10active_key_pairs\x18\x02 \x03(\v2(.teleport.recordingencryption.v1.KeyPairR\x0eactiveKeyPairsJ\x04\b\x01\x10\x02R\vactive_keys\"\x1b\n" +
 	"\x19RecordingEncryptionStatus\"\xba\x02\n" +
 	"\x13RecordingEncryption\x12\x12\n" +
 	"\x04kind\x18\x01 \x01(\tR\x04kind\x12\x19\n" +
@@ -303,7 +286,7 @@ func file_teleport_recordingencryption_v1_recording_encryption_proto_rawDescGZIP
 
 var file_teleport_recordingencryption_v1_recording_encryption_proto_msgTypes = make([]protoimpl.MessageInfo, 4)
 var file_teleport_recordingencryption_v1_recording_encryption_proto_goTypes = []any{
-	(*WrappedKey)(nil),                // 0: teleport.recordingencryption.v1.WrappedKey
+	(*KeyPair)(nil),                   // 0: teleport.recordingencryption.v1.KeyPair
 	(*RecordingEncryptionSpec)(nil),   // 1: teleport.recordingencryption.v1.RecordingEncryptionSpec
 	(*RecordingEncryptionStatus)(nil), // 2: teleport.recordingencryption.v1.RecordingEncryptionStatus
 	(*RecordingEncryption)(nil),       // 3: teleport.recordingencryption.v1.RecordingEncryption
@@ -311,17 +294,16 @@ var file_teleport_recordingencryption_v1_recording_encryption_proto_goTypes = []
 	(*v1.Metadata)(nil),               // 5: teleport.header.v1.Metadata
 }
 var file_teleport_recordingencryption_v1_recording_encryption_proto_depIdxs = []int32{
-	4, // 0: teleport.recordingencryption.v1.WrappedKey.recording_encryption_pair:type_name -> types.EncryptionKeyPair
-	4, // 1: teleport.recordingencryption.v1.WrappedKey.key_encryption_pair:type_name -> types.EncryptionKeyPair
-	0, // 2: teleport.recordingencryption.v1.RecordingEncryptionSpec.active_keys:type_name -> teleport.recordingencryption.v1.WrappedKey
-	5, // 3: teleport.recordingencryption.v1.RecordingEncryption.metadata:type_name -> teleport.header.v1.Metadata
-	1, // 4: teleport.recordingencryption.v1.RecordingEncryption.spec:type_name -> teleport.recordingencryption.v1.RecordingEncryptionSpec
-	2, // 5: teleport.recordingencryption.v1.RecordingEncryption.status:type_name -> teleport.recordingencryption.v1.RecordingEncryptionStatus
-	6, // [6:6] is the sub-list for method output_type
-	6, // [6:6] is the sub-list for method input_type
-	6, // [6:6] is the sub-list for extension type_name
-	6, // [6:6] is the sub-list for extension extendee
-	0, // [0:6] is the sub-list for field type_name
+	4, // 0: teleport.recordingencryption.v1.KeyPair.key_pair:type_name -> types.EncryptionKeyPair
+	0, // 1: teleport.recordingencryption.v1.RecordingEncryptionSpec.active_key_pairs:type_name -> teleport.recordingencryption.v1.KeyPair
+	5, // 2: teleport.recordingencryption.v1.RecordingEncryption.metadata:type_name -> teleport.header.v1.Metadata
+	1, // 3: teleport.recordingencryption.v1.RecordingEncryption.spec:type_name -> teleport.recordingencryption.v1.RecordingEncryptionSpec
+	2, // 4: teleport.recordingencryption.v1.RecordingEncryption.status:type_name -> teleport.recordingencryption.v1.RecordingEncryptionStatus
+	5, // [5:5] is the sub-list for method output_type
+	5, // [5:5] is the sub-list for method input_type
+	5, // [5:5] is the sub-list for extension type_name
+	5, // [5:5] is the sub-list for extension extendee
+	0, // [0:5] is the sub-list for field type_name
 }
 
 func init() { file_teleport_recordingencryption_v1_recording_encryption_proto_init() }
diff --git a/api/proto/teleport/legacy/types/types.proto b/api/proto/teleport/legacy/types/types.proto
index 8a86be7fa23d2..96e841a87eb93 100644
--- a/api/proto/teleport/legacy/types/types.proto
+++ b/api/proto/teleport/legacy/types/types.proto
@@ -1269,9 +1269,9 @@ message EncryptionKeyPair {
   uint32 hash = 4 [(gogoproto.jsontag) = "hash,omitempty"];
 }
 
-// AgeEncryptionKey is a Bech32 encoded age X25519 public key.
+// A public key to be used as a recipient during age encryption of session recordings.
 message AgeEncryptionKey {
-  // PublicKey is a Bech32 encoded age X25519 public key.
+  // A PEM encoded public key used for key wrapping during age encryption. Expected to be RSA 4096.
   bytes public_key = 1 [(gogoproto.jsontag) = "public_key"];
 }
 
diff --git a/api/proto/teleport/recordingencryption/v1/recording_encryption.proto b/api/proto/teleport/recordingencryption/v1/recording_encryption.proto
index 5ee6ef9f0f014..0999f68168e6b 100644
--- a/api/proto/teleport/recordingencryption/v1/recording_encryption.proto
+++ b/api/proto/teleport/recordingencryption/v1/recording_encryption.proto
@@ -21,26 +21,22 @@ import "teleport/legacy/types/types.proto";
 
 option go_package = "github.com/gravitational/teleport/api/gen/proto/go/teleport/recordingencryption/v1;recordingencryptionv1";
 
-// WrappedKey wraps the private key of a recording encryption key pair using a
-// separate asymmetric keypair.
-message WrappedKey {
-  // RecordingEncryptionPair is the asymmetric keypair used with age to encrypt
-  // and decrypt filekeys. The private key is encrypted using the KeyEncryptionPair's
-  // public key and has to be decrypted before recording decryption operations can
-  // be fulfilled.
-  types.EncryptionKeyPair recording_encryption_pair = 1;
-  // KeyEncryptionPair is the asymmetric keypair used to wrap (encrypt) the
-  // RecordingEncryptionPair's private key.
-  types.EncryptionKeyPair key_encryption_pair = 2;
+// A key pair used with age to wrap and unwrap file keys for session recording encryption.
+message KeyPair {
+  // A key pair used with age to wrap and unwrap file keys for session recording encryption.
+  types.EncryptionKeyPair key_pair = 1;
 }
 
 // RecordingEncryptionSpec contains the active key set for encrypted session recording.
 message RecordingEncryptionSpec {
-  // AciveKeys is a list of active, wrapped X25519 keypairs. The unique set of RecordingEncryptionPair
-  // public keys are used as recipients during age encryption of session recordings. This
-  // allows any active private key to be used during decryption which guards against recordings
-  // being inaccessible to auth servers waiting for their key to rotate.
-  repeated WrappedKey active_keys = 1;
+  reserved 1; // active_keys
+  reserved "active_keys";
+
+  // A list of active key pairs used for session recording encryption. The unique set of
+  // active public keys are used as recipients during age encryption. This allows any
+  // active private key to be used during decryption which guards against recordings being
+  // inaccessible to auth servers waiting for key rotation.
+  repeated KeyPair active_key_pairs = 2;
 }
 
 // RecordingEncryptionStatus contains the status of the RecordingEncryption resource.
diff --git a/api/types/sessionrecording.go b/api/types/sessionrecording.go
index a3138a368dd1b..ad1fc5af60c8f 100644
--- a/api/types/sessionrecording.go
+++ b/api/types/sessionrecording.go
@@ -59,6 +59,7 @@ type SessionRecordingConfig interface {
 
 	// Clone returns a copy of the resource.
 	Clone() SessionRecordingConfig
+
 	// CheckAndSetDefaults verifies the constraints for a SessionRecordingConfig
 	CheckAndSetDefaults() error
 }
diff --git a/api/types/types.pb.go b/api/types/types.pb.go
index cb50f7414baca..bdba3fc9b1c82 100644
--- a/api/types/types.pb.go
+++ b/api/types/types.pb.go
@@ -4413,9 +4413,9 @@ func (m *EncryptionKeyPair) XXX_DiscardUnknown() {
 
 var xxx_messageInfo_EncryptionKeyPair proto.InternalMessageInfo
 
-// AgeEncryptionKey is a Bech32 encoded age X25519 public key.
+// A public key to be used as a recipient during age encryption of session recordings.
 type AgeEncryptionKey struct {
-	// PublicKey is a Bech32 encoded age X25519 public key.
+	// A PEM encoded public key used for key wrapping during age encryption. Expected to be RSA 4096.
 	PublicKey            []byte   `protobuf:"bytes,1,opt,name=public_key,json=publicKey,proto3" json:"public_key"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
diff --git a/docs/pages/reference/terraform-provider/data-sources/session_recording_config.mdx b/docs/pages/reference/terraform-provider/data-sources/session_recording_config.mdx
index 7ee2e550a7522..f82e132b02448 100644
--- a/docs/pages/reference/terraform-provider/data-sources/session_recording_config.mdx
+++ b/docs/pages/reference/terraform-provider/data-sources/session_recording_config.mdx
@@ -89,5 +89,5 @@ Optional:
 
 Optional:
 
-- `public_key` (String) PublicKey is a Bech32 encoded age X25519 public key.
+- `public_key` (String) A PEM encoded public key used for key wrapping during age encryption. Expected to be RSA 4096.
 
diff --git a/docs/pages/reference/terraform-provider/resources/session_recording_config.mdx b/docs/pages/reference/terraform-provider/resources/session_recording_config.mdx
index 30d0a29ca88cd..17e0f9c50339a 100644
--- a/docs/pages/reference/terraform-provider/resources/session_recording_config.mdx
+++ b/docs/pages/reference/terraform-provider/resources/session_recording_config.mdx
@@ -111,4 +111,4 @@ Optional:
 
 Optional:
 
-- `public_key` (String) PublicKey is a Bech32 encoded age X25519 public key.
+- `public_key` (String) A PEM encoded public key used for key wrapping during age encryption. Expected to be RSA 4096.
diff --git a/integrations/terraform/tfschema/types_terraform.go b/integrations/terraform/tfschema/types_terraform.go
index 28b150f390837..ba3ba2840c0d6 100644
--- a/integrations/terraform/tfschema/types_terraform.go
+++ b/integrations/terraform/tfschema/types_terraform.go
@@ -2157,7 +2157,7 @@ func GenSchemaSessionRecordingConfigV2(ctx context.Context) (github_com_hashicor
 		"status": {
 			Attributes: github_com_hashicorp_terraform_plugin_framework_tfsdk.SingleNestedAttributes(map[string]github_com_hashicorp_terraform_plugin_framework_tfsdk.Attribute{"encryption_keys": {
 				Attributes: github_com_hashicorp_terraform_plugin_framework_tfsdk.ListNestedAttributes(map[string]github_com_hashicorp_terraform_plugin_framework_tfsdk.Attribute{"public_key": {
-					Description: "PublicKey is a Bech32 encoded age X25519 public key.",
+					Description: "A PEM encoded public key used for key wrapping during age encryption. Expected to be RSA 4096.",
 					Optional:    true,
 					Type:        github_com_hashicorp_terraform_plugin_framework_types.StringType,
 				}}),
diff --git a/lib/auth/clusterconfig/clusterconfigv1/service.go b/lib/auth/clusterconfig/clusterconfigv1/service.go
index 029276bcd09f5..fef0a01fa5888 100644
--- a/lib/auth/clusterconfig/clusterconfigv1/service.go
+++ b/lib/auth/clusterconfig/clusterconfigv1/service.go
@@ -738,7 +738,7 @@ func (s *Service) CreateSessionRecordingConfig(ctx context.Context, cfg types.Se
 		return nil, trace.AccessDenied("this request can be only executed by an auth server")
 	}
 
-	if err := services.ValidateSessionRecordingConfig(cfg); err != nil {
+	if err := services.ValidateSessionRecordingConfig(cfg, s.signatureAlgorithmSuiteParams.FIPS, s.signatureAlgorithmSuiteParams.Cloud); err != nil {
 		return nil, trace.Wrap(err)
 	}
 
@@ -776,9 +776,10 @@ func (s *Service) UpdateSessionRecordingConfig(ctx context.Context, req *cluster
 
 	req.SessionRecordingConfig.SetOrigin(types.OriginDynamic)
 
-	if err := services.ValidateSessionRecordingConfig(req.SessionRecordingConfig); err != nil {
+	if err := services.ValidateSessionRecordingConfig(req.SessionRecordingConfig, s.signatureAlgorithmSuiteParams.FIPS, s.signatureAlgorithmSuiteParams.Cloud); err != nil {
 		return nil, trace.Wrap(err)
 	}
+
 	updated, err := s.backend.UpdateSessionRecordingConfig(ctx, req.SessionRecordingConfig)
 
 	if err := s.emitter.EmitAuditEvent(ctx, &apievents.SessionRecordingConfigUpdate{
@@ -825,9 +826,10 @@ func (s *Service) UpsertSessionRecordingConfig(ctx context.Context, req *cluster
 
 	req.SessionRecordingConfig.SetOrigin(types.OriginDynamic)
 
-	if err := services.ValidateSessionRecordingConfig(req.SessionRecordingConfig); err != nil {
+	if err := services.ValidateSessionRecordingConfig(req.SessionRecordingConfig, s.signatureAlgorithmSuiteParams.FIPS, s.signatureAlgorithmSuiteParams.Cloud); err != nil {
 		return nil, trace.Wrap(err)
 	}
+
 	upserted, err := s.backend.UpsertSessionRecordingConfig(ctx, req.SessionRecordingConfig)
 
 	if err := s.emitter.EmitAuditEvent(ctx, &apievents.SessionRecordingConfigUpdate{
diff --git a/lib/auth/keystore/aws_kms.go b/lib/auth/keystore/aws_kms.go
index 6c5d5b99c08cc..a09507abb7681 100644
--- a/lib/auth/keystore/aws_kms.go
+++ b/lib/auth/keystore/aws_kms.go
@@ -54,9 +54,10 @@ import (
 )
 
 const (
-	awskmsPrefix  = "awskms:"
-	clusterTagKey = "TeleportCluster"
-	awsOAEPHash   = crypto.SHA256
+	awskmsPrefix           = "awskms:"
+	clusterTagKey          = "TeleportCluster"
+	encryptedClusterTagKey = "TeleportClusterEncryption"
+	awsOAEPHash            = crypto.SHA256
 
 	pendingKeyBaseRetryInterval = time.Second / 2
 	pendingKeyMaxRetryInterval  = 4 * time.Second
@@ -131,7 +132,7 @@ func newAWSKMSKeystore(ctx context.Context, cfg *servicecfg.AWSKMSConfig, opts *
 
 	tags := cfg.Tags
 	if tags == nil {
-		tags = make(map[string]string, 1)
+		tags = make(map[string]string, 2)
 	}
 	if _, ok := tags[clusterTagKey]; !ok {
 		tags[clusterTagKey] = opts.ClusterName.GetClusterName()
@@ -196,6 +197,9 @@ func (a *awsKMSKeystore) generateKey(ctx context.Context, algorithm cryptosuites
 
 	tags := make([]kmstypes.Tag, 0, len(a.tags))
 	for k, v := range a.tags {
+		if k == clusterTagKey && usage == keyUsageDecrypt {
+			k = encryptedClusterTagKey
+		}
 		tags = append(tags, kmstypes.Tag{
 			TagKey:   aws.String(k),
 			TagValue: aws.String(v),
@@ -545,7 +549,14 @@ func (a *awsKMSKeystore) deleteUnusedKeys(ctx context.Context, activeKeys [][]by
 	if scopedKeyDeletion {
 		keyListFn = a.forEachOwnedKey
 	}
-	err := keyListFn(ctx, func(ctx context.Context, arn string) error {
+	err := keyListFn(ctx, func(ctx context.Context, arn string, tags []rgttypes.Tag) error {
+		if slices.ContainsFunc(tags, func(tag rgttypes.Tag) bool {
+			return aws.ToString(tag.Key) == encryptedClusterTagKey
+		}) {
+			// do nothing for keys marked with delete prevention
+			return nil
+		}
+
 		key, err := keyIDFromArn(arn)
 		if err != nil {
 			return trace.Wrap(err)
@@ -645,7 +656,7 @@ func (a *awsKMSKeystore) deleteUnusedKeys(ctx context.Context, activeKeys [][]by
 
 // forEachKey calls fn with the AWS key ID of all keys in the AWS account and
 // region that would be returned by ListKeys. It may call fn concurrently.
-func (a *awsKMSKeystore) forEachKey(ctx context.Context, fn func(ctx context.Context, keyARN string) error) error {
+func (a *awsKMSKeystore) forEachKey(ctx context.Context, fn func(ctx context.Context, keyARN string, tags []rgttypes.Tag) error) error {
 	errGroup, ctx := errgroup.WithContext(ctx)
 	marker := ""
 	more := true
@@ -666,7 +677,7 @@ func (a *awsKMSKeystore) forEachKey(ctx context.Context, fn func(ctx context.Con
 		for _, keyEntry := range output.Keys {
 			keyID := aws.ToString(keyEntry.KeyArn)
 			errGroup.Go(func() error {
-				return trace.Wrap(fn(ctx, keyID))
+				return trace.Wrap(fn(ctx, keyID, nil))
 			})
 		}
 	}
@@ -675,7 +686,7 @@ func (a *awsKMSKeystore) forEachKey(ctx context.Context, fn func(ctx context.Con
 
 // forEachOwnedKey calls fn with the AWS key ID of all owned keys in the AWS account and
 // region that would be returned by GetResources. It may call fn concurrently.
-func (a *awsKMSKeystore) forEachOwnedKey(ctx context.Context, fn func(ctx context.Context, keyARN string) error) error {
+func (a *awsKMSKeystore) forEachOwnedKey(ctx context.Context, fn func(ctx context.Context, keyARN string, tags []rgttypes.Tag) error) error {
 	const maxGoRoutines = 5
 	errGroup, ctx := errgroup.WithContext(ctx)
 	errGroup.SetLimit(maxGoRoutines)
@@ -705,7 +716,7 @@ func (a *awsKMSKeystore) forEachOwnedKey(ctx context.Context, fn func(ctx contex
 		for _, keyEntry := range output.ResourceTagMappingList {
 			keyID := aws.ToString(keyEntry.ResourceARN)
 			errGroup.Go(func() error {
-				return trace.Wrap(fn(ctx, keyID))
+				return trace.Wrap(fn(ctx, keyID, keyEntry.Tags))
 			})
 		}
 	}
diff --git a/lib/auth/keystore/gcp_kms.go b/lib/auth/keystore/gcp_kms.go
index 9b1e6797900d5..0cd36d1aebf51 100644
--- a/lib/auth/keystore/gcp_kms.go
+++ b/lib/auth/keystore/gcp_kms.go
@@ -45,6 +45,7 @@ import (
 const (
 	// GCP does not allow "." or "/" in labels
 	hostLabel                      = "teleport_auth_host"
+	encryptedHostLabel             = "teleport_encryption_auth_host"
 	gcpkmsPrefix                   = "gcpkms:"
 	gcpOAEPHash                    = crypto.SHA256
 	defaultGCPRequestTimeout       = 30 * time.Second
@@ -120,13 +121,18 @@ func (g *gcpKMSKeyStore) generateKey(ctx context.Context, algorithm cryptosuites
 	keyUUID := uuid.NewString()
 	g.log.InfoContext(ctx, "Creating new GCP KMS keypair.", "id", keyUUID, "algorithm", alg.String())
 
+	label := hostLabel
+	if usage == keyUsageDecrypt {
+		label = encryptedHostLabel
+	}
+
 	req := &kmspb.CreateCryptoKeyRequest{
 		Parent:      g.keyRing,
 		CryptoKeyId: keyUUID,
 		CryptoKey: &kmspb.CryptoKey{
 			Purpose: usage.toGCP(),
 			Labels: map[string]string{
-				hostLabel: g.hostUUID,
+				label: g.hostUUID,
 			},
 			VersionTemplate: &kmspb.CryptoKeyVersionTemplate{
 				ProtectionLevel: g.protectionLevel,
diff --git a/lib/auth/keystore/keystore_test.go b/lib/auth/keystore/keystore_test.go
index f078acea9b00e..d9063bd1f9d41 100644
--- a/lib/auth/keystore/keystore_test.go
+++ b/lib/auth/keystore/keystore_test.go
@@ -222,13 +222,17 @@ func TestBackends(t *testing.T) {
 				publicKeys[i] = signer.Public()
 			}
 
+			// generate an encryption key that should not be cleaned up
+			encryptionKey, decrypter, hash, err := backend.generateDecrypter(ctx, cryptosuites.RSA4096)
+			require.NoError(t, err)
+
 			// AWS KMS keystore will not delete any keys created in the past 5
 			// minutes.
 			pack.clock.Advance(6 * time.Minute)
 
 			// say that only the first key is in use, delete the rest
 			usedKeys := [][]byte{rawPrivateKeys[0]}
-			err := backend.deleteUnusedKeys(ctx, usedKeys)
+			err = backend.deleteUnusedKeys(ctx, usedKeys)
 			require.NoError(t, err, trace.DebugReport(err))
 
 			// make sure the first key is still good
@@ -237,6 +241,23 @@ func TestBackends(t *testing.T) {
 			_, err = signer.Sign(rand.Reader, messageHash[:], crypto.SHA256)
 			require.NoError(t, err)
 
+			// make sure the encryption key is still good
+			plaintext := "test message"
+			pubKey, ok := decrypter.Public().(*rsa.PublicKey)
+			require.True(t, ok, "expected *rsa.PublicKey, got %T", decrypter.Public())
+
+			cipher, err := rsa.EncryptOAEP(hash.New(), rand.Reader, pubKey, []byte(plaintext), nil)
+			require.NoError(t, err)
+
+			decrypter, err = backend.getDecrypter(ctx, encryptionKey, decrypter.Public(), hash)
+			require.NoError(t, err)
+
+			decrypted, err := decrypter.Decrypt(rand.Reader, cipher, &rsa.OAEPOptions{
+				Hash: hash,
+			})
+			require.NoError(t, err)
+			require.Equal(t, plaintext, string(decrypted))
+
 			// make sure all other keys are deleted
 			for i := 1; i < numKeys; i++ {
 				signer, err := backend.getSigner(ctx, rawPrivateKeys[i], publicKeys[0])
diff --git a/lib/auth/keystore/pkcs11.go b/lib/auth/keystore/pkcs11.go
index 024228b09ee8c..e02cc29104485 100644
--- a/lib/auth/keystore/pkcs11.go
+++ b/lib/auth/keystore/pkcs11.go
@@ -44,6 +44,10 @@ import (
 
 var pkcs11Prefix = []byte("pkcs11:")
 
+const (
+	recordingEncryptionHostID = "teleport_recording_encryption"
+)
+
 type pkcs11KeyStore struct {
 	ctx       *crypto11.Context
 	hostUUID  string
@@ -198,6 +202,7 @@ func (p *pkcs11KeyStore) generateDecrypter(ctx context.Context, alg cryptosuites
 		return nil, nil, p.oaepHash, trace.Wrap(err)
 	}
 
+	id.HostID = recordingEncryptionHostID
 	rawTeleportID, err := id.marshal()
 	if err != nil {
 		return nil, nil, p.oaepHash, trace.Wrap(err)
@@ -210,7 +215,7 @@ func (p *pkcs11KeyStore) generateDecrypter(ctx context.Context, alg cryptosuites
 
 	p.log.InfoContext(ctx, "Creating new HSM keypair.", "id", id, "algorithm", logutils.StringerAttr(alg))
 
-	label := []byte(p.hostUUID)
+	label := []byte(id.HostID)
 	switch alg {
 	case cryptosuites.RSA4096:
 		decrypter, err := p.generateRSA4096(rawPKCS11ID, label)
@@ -329,6 +334,10 @@ func (p *pkcs11KeyStore) findDecryptersByLabel(ctx context.Context, label *types
 	return decrypters, nil
 }
 
+func (p *pkcs11KeyStore) checkAccessibleHostID(hostID string) bool {
+	return hostID == recordingEncryptionHostID || hostID == p.hostUUID
+}
+
 func (p *pkcs11KeyStore) getSignerWithoutPublicKey(ctx context.Context, rawKey []byte) (crypto.Signer, error) {
 	if t := keyType(rawKey); t != types.PrivateKeyType_PKCS11 {
 		return nil, trace.BadParameter("pkcs11KeyStore cannot get signer for key type %s", t.String())
@@ -337,14 +346,14 @@ func (p *pkcs11KeyStore) getSignerWithoutPublicKey(ctx context.Context, rawKey [
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
-	if keyID.HostID != p.hostUUID {
+	if !p.checkAccessibleHostID(keyID.HostID) {
 		return nil, trace.NotFound("given pkcs11 key is for host: %q, but this host is: %q", keyID.HostID, p.hostUUID)
 	}
 	pkcs11ID, err := keyID.pkcs11Key(p.isYubiHSM)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
-	signer, err := p.ctx.FindKeyPair(pkcs11ID, []byte(p.hostUUID))
+	signer, err := p.ctx.FindKeyPair(pkcs11ID, []byte(keyID.HostID))
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
@@ -367,7 +376,7 @@ func (p *pkcs11KeyStore) canUseKey(ctx context.Context, raw []byte, keyType type
 		return false, trace.Wrap(err)
 	}
 
-	return keyID.HostID == p.hostUUID, nil
+	return p.checkAccessibleHostID(keyID.HostID), nil
 }
 
 // deleteKey deletes the given key from the HSM
@@ -376,7 +385,7 @@ func (p *pkcs11KeyStore) deleteKey(_ context.Context, rawKey []byte) error {
 	if err != nil {
 		return trace.Wrap(err)
 	}
-	if keyID.HostID != p.hostUUID {
+	if !p.checkAccessibleHostID(keyID.HostID) {
 		return trace.NotFound("pkcs11 key is for different host")
 	}
 	pkcs11ID, err := keyID.pkcs11Key(p.isYubiHSM)
@@ -413,7 +422,7 @@ func (p *pkcs11KeyStore) deleteUnusedKeys(ctx context.Context, activeKeys [][]by
 		if err != nil {
 			return trace.Wrap(err)
 		}
-		if keyID.HostID != p.hostUUID {
+		if !p.checkAccessibleHostID(keyID.HostID) {
 			// This key was labeled with a foreign host UUID, it is likely not
 			// present on the attached HSM and definitely will not be returned
 			// by FindKeyPairs below which queries by host UUID.
diff --git a/lib/auth/recordingencryption/manager.go b/lib/auth/recordingencryption/manager.go
index d6fa97c453cf7..f05e998ed118e 100644
--- a/lib/auth/recordingencryption/manager.go
+++ b/lib/auth/recordingencryption/manager.go
@@ -121,7 +121,7 @@ func (m *Manager) CreateSessionRecordingConfig(ctx context.Context, cfg types.Se
 				return err
 			}
 
-			_ = cfg.SetEncryptionKeys(getAgeEncryptionKeys(encryption.GetSpec().ActiveKeys))
+			_ = cfg.SetEncryptionKeys(getAgeEncryptionKeys(encryption.GetSpec().ActiveKeyPairs))
 		}
 
 		sessionRecordingConfig, err = m.ClusterConfigurationInternal.CreateSessionRecordingConfig(ctx, cfg)
@@ -147,7 +147,7 @@ func (m *Manager) UpdateSessionRecordingConfig(ctx context.Context, cfg types.Se
 				return err
 			}
 
-			_ = cfg.SetEncryptionKeys(getAgeEncryptionKeys(encryption.GetSpec().ActiveKeys))
+			_ = cfg.SetEncryptionKeys(getAgeEncryptionKeys(encryption.GetSpec().ActiveKeyPairs))
 		}
 
 		sessionRecordingConfig, err = m.ClusterConfigurationInternal.UpdateSessionRecordingConfig(ctx, cfg)
@@ -173,7 +173,7 @@ func (m *Manager) UpsertSessionRecordingConfig(ctx context.Context, cfg types.Se
 				return err
 			}
 
-			_ = cfg.SetEncryptionKeys(getAgeEncryptionKeys(encryption.GetSpec().ActiveKeys))
+			_ = cfg.SetEncryptionKeys(getAgeEncryptionKeys(encryption.GetSpec().ActiveKeyPairs))
 		}
 
 		sessionRecordingConfig, err = m.ClusterConfigurationInternal.UpsertSessionRecordingConfig(ctx, cfg)
@@ -233,22 +233,22 @@ func (m *Manager) ensureManualEncryptionKeys(manualKeyCfg types.ManualKeyManagem
 		}
 	})
 
-	var encryptionKeys []*recordingencryptionv1.WrappedKey
+	var encryptionKeys []*recordingencryptionv1.KeyPair
 	for _, decrypter := range activeDecrypters {
 		pubKey, err := keys.MarshalPublicKey(decrypter.Public())
 		if err != nil {
 			return nil, trace.Wrap(err)
 		}
 
-		encryptionKeys = append(encryptionKeys, &recordingencryptionv1.WrappedKey{
-			RecordingEncryptionPair: &types.EncryptionKeyPair{
+		encryptionKeys = append(encryptionKeys, &recordingencryptionv1.KeyPair{
+			KeyPair: &types.EncryptionKeyPair{
 				PublicKey: pubKey,
 			},
 		})
 	}
 	return &recordingencryptionv1.RecordingEncryption{
 		Spec: &recordingencryptionv1.RecordingEncryptionSpec{
-			ActiveKeys: encryptionKeys,
+			ActiveKeyPairs: encryptionKeys,
 		},
 	}, nil
 }
@@ -274,12 +274,12 @@ func (m *Manager) ensureRecordingEncryptionKey(ctx context.Context, encryptionCf
 		persistFn = m.RecordingEncryption.CreateRecordingEncryption
 	}
 
-	activeKeys := encryption.GetSpec().ActiveKeys
-	if len(activeKeys) > 0 {
-		for _, key := range activeKeys {
+	activePairs := encryption.GetSpec().ActiveKeyPairs
+	if len(activePairs) > 0 {
+		for _, pair := range activePairs {
 			// fetch the decrypter to ensure we have access to it
-			if _, err := m.keyStore.GetDecrypter(ctx, key.RecordingEncryptionPair); err != nil {
-				fp, _ := fingerprintPEM(key.RecordingEncryptionPair.PublicKey)
+			if _, err := m.keyStore.GetDecrypter(ctx, pair.KeyPair); err != nil {
+				fp, _ := fingerprintPEM(pair.KeyPair.PublicKey)
 				m.logger.DebugContext(ctx, "key not accessible", "fingerprint", fp)
 				continue
 			}
@@ -295,10 +295,10 @@ func (m *Manager) ensureRecordingEncryptionKey(ctx context.Context, encryptionCf
 		return nil, trace.Wrap(err, "generating wrapping key")
 	}
 
-	wrappedKey := recordingencryptionv1.WrappedKey{
-		RecordingEncryptionPair: encryptionPair,
+	wrappedKey := recordingencryptionv1.KeyPair{
+		KeyPair: encryptionPair,
 	}
-	encryption.Spec.ActiveKeys = []*recordingencryptionv1.WrappedKey{&wrappedKey}
+	encryption.Spec.ActiveKeyPairs = []*recordingencryptionv1.KeyPair{&wrappedKey}
 	encryption, err = persistFn(ctx, encryption)
 	if err != nil {
 		return nil, trace.Wrap(err)
@@ -346,13 +346,13 @@ func (m *Manager) UnwrapKey(ctx context.Context, in UnwrapInput) ([]byte, error)
 	}
 
 	// TODO (eriktate): search rotated keys as well once rotation is implemented
-	activeKeys := encryption.GetSpec().ActiveKeys
-	for _, key := range activeKeys {
-		if key.GetRecordingEncryptionPair() == nil {
+	activePairs := encryption.GetSpec().ActiveKeyPairs
+	for _, key := range activePairs {
+		if key.GetKeyPair() == nil {
 			continue
 		}
 
-		activeFP, err := fingerprintPEM(key.RecordingEncryptionPair.PublicKey)
+		activeFP, err := fingerprintPEM(key.KeyPair.PublicKey)
 		if err != nil {
 			m.logger.ErrorContext(ctx, "failed to fingerprint active public key", "error", err)
 			continue
@@ -362,7 +362,7 @@ func (m *Manager) UnwrapKey(ctx context.Context, in UnwrapInput) ([]byte, error)
 			continue
 		}
 
-		decrypter, err := m.keyStore.GetDecrypter(ctx, key.RecordingEncryptionPair)
+		decrypter, err := m.keyStore.GetDecrypter(ctx, key.KeyPair)
 		if err != nil {
 			continue
 		}
@@ -487,7 +487,7 @@ func (m *Manager) resolveRecordingEncryption(ctx context.Context, shouldRetryFn
 				return trace.Wrap(err)
 			}
 
-			if sessionRecordingConfig.SetEncryptionKeys(getAgeEncryptionKeys(encryption.GetSpec().ActiveKeys)) {
+			if sessionRecordingConfig.SetEncryptionKeys(getAgeEncryptionKeys(encryption.GetSpec().ActiveKeyPairs)) {
 				if _, err := m.ClusterConfigurationInternal.UpdateSessionRecordingConfig(ctx, sessionRecordingConfig); err != nil {
 					return trace.Wrap(err)
 				}
@@ -507,15 +507,15 @@ func (m *Manager) resolveRecordingEncryption(ctx context.Context, shouldRetryFn
 
 // getAgeEncryptionKeys returns an iterator of AgeEncryptionKeys from a list of WrappedKeys. This is for use in
 // populating the EncryptionKeys field of SessionRecordingConfigStatus.
-func getAgeEncryptionKeys(keys []*recordingencryptionv1.WrappedKey) iter.Seq[*types.AgeEncryptionKey] {
+func getAgeEncryptionKeys(keys []*recordingencryptionv1.KeyPair) iter.Seq[*types.AgeEncryptionKey] {
 	return func(yield func(*types.AgeEncryptionKey) bool) {
 		for _, key := range keys {
-			if key.RecordingEncryptionPair == nil {
+			if key.KeyPair == nil {
 				continue
 			}
 
 			if !yield(&types.AgeEncryptionKey{
-				PublicKey: key.RecordingEncryptionPair.PublicKey,
+				PublicKey: key.KeyPair.PublicKey,
 			}) {
 				return
 			}
diff --git a/lib/auth/recordingencryption/manager_test.go b/lib/auth/recordingencryption/manager_test.go
index 2ff377b65fa9b..c49753eddabe7 100644
--- a/lib/auth/recordingencryption/manager_test.go
+++ b/lib/auth/recordingencryption/manager_test.go
@@ -282,11 +282,11 @@ func TestCreateUpdateSessionRecordingConfig(t *testing.T) {
 
 	encryption, err := config.Backend.GetRecordingEncryption(ctx)
 	require.NoError(t, err)
-	activeKeys := encryption.GetSpec().GetActiveKeys()
-	require.Len(t, activeKeys, 1)
-	require.NotNil(t, activeKeys[0].RecordingEncryptionPair)
-	require.NotEmpty(t, activeKeys[0].RecordingEncryptionPair.PrivateKey)
-	require.NotEmpty(t, activeKeys[0].RecordingEncryptionPair.PublicKey)
+	activePairs := encryption.GetSpec().GetActiveKeyPairs()
+	require.Len(t, activePairs, 1)
+	require.NotNil(t, activePairs[0].KeyPair)
+	require.NotEmpty(t, activePairs[0].KeyPair.PrivateKey)
+	require.NotEmpty(t, activePairs[0].KeyPair.PublicKey)
 
 	// update should change nothing
 	src, err = manager.UpdateSessionRecordingConfig(ctx, src)
@@ -296,8 +296,8 @@ func TestCreateUpdateSessionRecordingConfig(t *testing.T) {
 
 	encryption, err = config.Backend.GetRecordingEncryption(ctx)
 	require.NoError(t, err)
-	newActiveKeys := encryption.GetSpec().GetActiveKeys()
-	require.ElementsMatch(t, newActiveKeys, activeKeys)
+	newActiveKeyPairs := encryption.GetSpec().GetActiveKeyPairs()
+	require.ElementsMatch(t, newActiveKeyPairs, activePairs)
 }
 
 func TestResolveRecordingEncryption(t *testing.T) {
@@ -327,22 +327,22 @@ func TestResolveRecordingEncryption(t *testing.T) {
 	// CASE: service A first evaluation initializes recording encryption resource
 	encryption, src, err := resolve(ctx, service, managerA)
 	require.NoError(t, err)
-	initialKeys := encryption.GetSpec().GetActiveKeys()
+	initialKeys := encryption.GetSpec().GetActiveKeyPairs()
 
 	require.Len(t, initialKeys, 1)
 	require.Len(t, src.GetEncryptionKeys(), 1)
 	key := initialKeys[0]
-	require.Equal(t, key.RecordingEncryptionPair.PublicKey, src.GetEncryptionKeys()[0].PublicKey)
-	require.NotNil(t, key.RecordingEncryptionPair)
+	require.Equal(t, key.KeyPair.PublicKey, src.GetEncryptionKeys()[0].PublicKey)
+	require.NotNil(t, key.KeyPair)
 
 	// CASE: service B should have access to the same key
 	encryption, src, err = resolve(ctx, service, managerB)
 	require.NoError(t, err)
 
-	activeKeys := encryption.GetSpec().ActiveKeys
+	activePairs := encryption.GetSpec().ActiveKeyPairs
 	require.Len(t, src.GetEncryptionKeys(), 1)
-	require.Equal(t, key.RecordingEncryptionPair.PublicKey, src.GetEncryptionKeys()[0].PublicKey)
-	require.ElementsMatch(t, initialKeys, activeKeys)
+	require.Equal(t, key.KeyPair.PublicKey, src.GetEncryptionKeys()[0].PublicKey)
+	require.ElementsMatch(t, initialKeys, activePairs)
 
 	// service C should error without access to the current key
 	_, _, err = resolve(ctx, service, managerC)
@@ -383,13 +383,13 @@ func TestResolveRecordingEncryptionConcurrent(t *testing.T) {
 	encryption, err := service.GetRecordingEncryption(ctx)
 	require.NoError(t, err)
 
-	activeKeys := encryption.GetSpec().ActiveKeys
+	activePairs := encryption.GetSpec().ActiveKeyPairs
 	// each service should share a single active key
-	require.Len(t, activeKeys, 1)
-	require.NotNil(t, activeKeys[0].RecordingEncryptionPair)
-	require.NotEmpty(t, activeKeys[0].RecordingEncryptionPair.PrivateKey)
-	require.NotEmpty(t, activeKeys[0].RecordingEncryptionPair.PublicKey)
-	require.Equal(t, types.PrivateKeyType_RAW, activeKeys[0].RecordingEncryptionPair.PrivateKeyType)
+	require.Len(t, activePairs, 1)
+	require.NotNil(t, activePairs[0].KeyPair)
+	require.NotEmpty(t, activePairs[0].KeyPair.PrivateKey)
+	require.NotEmpty(t, activePairs[0].KeyPair.PublicKey)
+	require.Equal(t, types.PrivateKeyType_RAW, activePairs[0].KeyPair.PrivateKeyType)
 }
 
 func TestUnwrapKey(t *testing.T) {
diff --git a/lib/cache/recording_encryption_test.go b/lib/cache/recording_encryption_test.go
index 3a161c0dc8274..1b16e310740da 100644
--- a/lib/cache/recording_encryption_test.go
+++ b/lib/cache/recording_encryption_test.go
@@ -35,7 +35,7 @@ func newRecordingEncryption() *recordingencryptionv1.RecordingEncryption {
 			Name: types.MetaNameRecordingEncryption,
 		},
 		Spec: &recordingencryptionv1.RecordingEncryptionSpec{
-			ActiveKeys: nil,
+			ActiveKeyPairs: nil,
 		},
 	}
 }
diff --git a/lib/config/configuration.go b/lib/config/configuration.go
index 0ea9c012088d1..2741fa544a45d 100644
--- a/lib/config/configuration.go
+++ b/lib/config/configuration.go
@@ -2592,6 +2592,10 @@ func Configure(clf *CommandLineFlags, cfg *servicecfg.Config, legacyAppFlags boo
 				!cfg.Auth.SessionRecordingConfig.GetProxyChecksHostKeys() {
 				return trace.BadParameter("non-FIPS compliant proxy settings: \"proxy_checks_host_keys\" must be true")
 			}
+
+			if err := services.ValidateSessionRecordingConfig(cfg.Auth.SessionRecordingConfig, clf.FIPS, modules.GetModules().Features().Cloud); err != nil {
+				return trace.Wrap(err)
+			}
 		}
 	}
 
diff --git a/lib/services/local/recording_encryption_test.go b/lib/services/local/recording_encryption_test.go
index 0f5420f02d1e1..c8ce0fc25466b 100644
--- a/lib/services/local/recording_encryption_test.go
+++ b/lib/services/local/recording_encryption_test.go
@@ -18,7 +18,6 @@ package local
 
 import (
 	"context"
-	"crypto"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -39,7 +38,7 @@ func TestRecordingEncryption(t *testing.T) {
 
 	initialEncryption := pb.RecordingEncryption{
 		Spec: &pb.RecordingEncryptionSpec{
-			ActiveKeys: nil,
+			ActiveKeyPairs: nil,
 		},
 	}
 
@@ -53,33 +52,28 @@ func TestRecordingEncryption(t *testing.T) {
 	encryption, err := service.GetRecordingEncryption(ctx)
 	require.NoError(t, err)
 
-	require.Empty(t, created.Spec.ActiveKeys)
-	require.Empty(t, encryption.Spec.ActiveKeys)
+	require.Empty(t, created.Spec.ActiveKeyPairs)
+	require.Empty(t, encryption.Spec.ActiveKeyPairs)
 
-	encryption.Spec.ActiveKeys = []*pb.WrappedKey{
+	encryption.Spec.ActiveKeyPairs = []*pb.KeyPair{
 		{
-			RecordingEncryptionPair: &types.EncryptionKeyPair{
+			KeyPair: &types.EncryptionKeyPair{
 				PrivateKey: []byte("recording encryption private"),
 				PublicKey:  []byte("recording encryption public"),
 				Hash:       0,
 			},
-			KeyEncryptionPair: &types.EncryptionKeyPair{
-				PrivateKey: []byte("key encryption private"),
-				PublicKey:  []byte("key encryption public"),
-				Hash:       uint32(crypto.SHA256),
-			},
 		},
 	}
 
 	updated, err := service.UpdateRecordingEncryption(ctx, encryption)
 	require.NoError(t, err)
-	require.Len(t, updated.Spec.ActiveKeys, 1)
-	require.EqualExportedValues(t, encryption.Spec.ActiveKeys[0], updated.Spec.ActiveKeys[0])
+	require.Len(t, updated.Spec.ActiveKeyPairs, 1)
+	require.EqualExportedValues(t, encryption.Spec.ActiveKeyPairs[0], updated.Spec.ActiveKeyPairs[0])
 
 	encryption, err = service.GetRecordingEncryption(ctx)
 	require.NoError(t, err)
-	require.Len(t, encryption.Spec.ActiveKeys, 1)
-	require.EqualExportedValues(t, updated.Spec.ActiveKeys[0], encryption.Spec.ActiveKeys[0])
+	require.Len(t, encryption.Spec.ActiveKeyPairs, 1)
+	require.EqualExportedValues(t, updated.Spec.ActiveKeyPairs[0], encryption.Spec.ActiveKeyPairs[0])
 
 	err = service.DeleteRecordingEncryption(ctx)
 	require.NoError(t, err)
diff --git a/lib/services/sessionrecording.go b/lib/services/sessionrecording.go
index 8d738781c40a9..c808d1048f575 100644
--- a/lib/services/sessionrecording.go
+++ b/lib/services/sessionrecording.go
@@ -97,8 +97,11 @@ const (
 	KeyTypeSoftware = "software"
 )
 
+var errRecordingEncryptionWithFIPS = &trace.BadParameterError{Message: `non-FIPS compliant session recording setting: "encryption" must not be enabled`}
+var errManualKeyManagementInCloud = &trace.BadParameterError{Message: `"manual_key_management" configuration is unsupported in Teleport Cloud`}
+
 // ValidateSessionRecordingConfig checks that the state of a [SessionRecordingConfig] meets constraints.
-func ValidateSessionRecordingConfig(cfg types.SessionRecordingConfig) error {
+func ValidateSessionRecordingConfig(cfg types.SessionRecordingConfig, fips, cloud bool) error {
 	if !slices.Contains(types.SessionRecordingModes, cfg.GetMode()) {
 		return trace.BadParameter("session recording mode must be one of %v; got %q", strings.Join(types.SessionRecordingModes, ","), cfg.GetMode())
 	}
@@ -108,11 +111,19 @@ func ValidateSessionRecordingConfig(cfg types.SessionRecordingConfig) error {
 		return nil
 	}
 
+	if fips && encryptionCfg.Enabled {
+		return trace.Wrap(errRecordingEncryptionWithFIPS)
+	}
+
 	manualKeyManagement := encryptionCfg.ManualKeyManagement
 	if manualKeyManagement == nil || !manualKeyManagement.Enabled {
 		return nil
 	}
 
+	if cloud {
+		return trace.Wrap(errManualKeyManagementInCloud)
+	}
+
 	if len(manualKeyManagement.ActiveKeys) == 0 {
 		return trace.BadParameter("at least one active key must be configured when using manually managed encryption keys")
 	}
diff --git a/lib/services/sessionrecording_test.go b/lib/services/sessionrecording_test.go
index 7a8281c745cd2..afbc5f48ebdb4 100644
--- a/lib/services/sessionrecording_test.go
+++ b/lib/services/sessionrecording_test.go
@@ -31,6 +31,9 @@ func TestValidateSessionRecordingConfig(t *testing.T) {
 	cases := []struct {
 		name      string
 		spec      types.SessionRecordingConfigSpecV2
+		params    types.SignatureAlgorithmSuiteParams
+		cloud     bool
+		fips      bool
 		expectErr error
 	}{
 		{
@@ -38,6 +41,8 @@ func TestValidateSessionRecordingConfig(t *testing.T) {
 			spec: types.SessionRecordingConfigSpecV2{
 				Mode: "node",
 			},
+			fips:      true,
+			cloud:     true,
 			expectErr: nil,
 		},
 		{
@@ -76,6 +81,31 @@ func TestValidateSessionRecordingConfig(t *testing.T) {
 			},
 			expectErr: trace.BadParameter("session recording mode must be one of %v; got %q", strings.Join(types.SessionRecordingModes, ","), "invalid"),
 		},
+		{
+			name: "invalid config: encryption with FIPS",
+			spec: types.SessionRecordingConfigSpecV2{
+				Mode: "node",
+				Encryption: &types.SessionRecordingEncryptionConfig{
+					Enabled: true,
+				},
+			},
+			fips:      true,
+			expectErr: trace.BadParameter("non-FIPS compliant session recording setting"),
+		},
+		{
+			name: "invalid config: manual encryption in cloud",
+			spec: types.SessionRecordingConfigSpecV2{
+				Mode: "node",
+				Encryption: &types.SessionRecordingEncryptionConfig{
+					Enabled: true,
+					ManualKeyManagement: &types.ManualKeyManagementConfig{
+						Enabled: true,
+					},
+				},
+			},
+			cloud:     true,
+			expectErr: trace.BadParameter(`"manual_key_management" configuration is unsupported in Teleport Cloud`),
+		},
 		{
 			name: "invalid config: manual encryption without active keys",
 			spec: types.SessionRecordingConfigSpecV2{
@@ -137,7 +167,7 @@ func TestValidateSessionRecordingConfig(t *testing.T) {
 
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
-			err := services.ValidateSessionRecordingConfig(&types.SessionRecordingConfigV2{Spec: c.spec})
+			err := services.ValidateSessionRecordingConfig(&types.SessionRecordingConfigV2{Spec: c.spec}, c.fips, c.cloud)
 			if c.expectErr == nil {
 				require.NoError(t, err)
 			} else {