feat(backup_service): support backup encryption for manual backups

Author: ulya-sidorina

cmd/ydbcp/main.go

@@ -210,7 +210,7 @@ func main() {
 		xlog.Info(ctx, "Created TtlWatcher")
 	}
 
-	backupScheduleHandler := handlers.NewBackupScheduleHandler(queries.NewWriteTableQuery, clockwork.NewRealClock())
+	backupScheduleHandler := handlers.NewBackupScheduleHandler(queries.NewWriteTableQuery, clockwork.NewRealClock(), configInstance.FeatureFlags)
 
 	schedule_watcher.NewScheduleWatcher(
 		ctx, &wg, configInstance.GetProcessorIntervalSeconds(), dbConnector,

internal/backup_operations/make_backup.go

@@ -2,6 +2,7 @@ package backup_operations
 
 import (
 	"context"
+	"crypto/rand"
 	"errors"
 	"fmt"
 	"github.com/jonboulle/clockwork"
@@ -35,6 +36,7 @@ type MakeBackupInternalRequest struct {
 	ScheduleID           *string
 	Ttl                  *time.Duration
 	ParentOperationID    *string
+	EncryptionSettings   *pb.EncryptionSettings
 }
 
 func FromBackupSchedule(schedule *types.BackupSchedule) MakeBackupInternalRequest {
@@ -62,6 +64,7 @@ func FromTBWROperation(tbwr *types.TakeBackupWithRetryOperation) MakeBackupInter
 		ScheduleID:           tbwr.ScheduleID,
 		Ttl:                  tbwr.Ttl,
 		ParentOperationID:    &tbwr.ID,
+		EncryptionSettings:   tbwr.EncryptionSettings,
 	}
 }
 
@@ -274,6 +277,34 @@ func IsEmptyBackup(backup *types.Backup) bool {
 	return backup.Size == 0 && backup.S3Endpoint == ""
 }
 
+func GetEncryptionParams(settings *pb.EncryptionSettings) ([]byte, string, error) {
+	var algorithm string
+	var length int
+
+	switch settings.Algorithm {
+	case pb.EncryptionSettings_UNSPECIFIED:
+	case pb.EncryptionSettings_AES_128_GCM:
+		algorithm = "AES-128-GCM"
+		length = 16
+		break
+	case pb.EncryptionSettings_AES_256_GCM:
+		algorithm = "AES-256-GCM"
+		length = 32
+		break
+	case pb.EncryptionSettings_CHACHA20_POLY1305:
+		algorithm = "ChaCha20-Poly1305"
+		length = 32
+		break
+	}
+
+	dek := make([]byte, length)
+	_, err := rand.Read(dek)
+	if err != nil {
+		return nil, "", err
+	}
+	return dek, algorithm, nil
+}
+
 func MakeBackup(
 	ctx context.Context,
 	clientConn client.ClientConnector,
@@ -350,6 +381,18 @@ func MakeBackup(
 		S3ForcePathStyle:  s3.S3ForcePathStyle,
 	}
 
+	if req.EncryptionSettings != nil && featureFlags.EnableBackupEncryption {
+		dek, algorithm, err := GetEncryptionParams(req.EncryptionSettings)
+		if err != nil {
+			return nil, nil, err
+		}
+
+		s3Settings.EncryptionKey = dek
+		s3Settings.EncryptionAlgorithm = algorithm
+		// TODO: encrypt the DEK using the specified KEK
+		// TODO: stores the encrypted DEK in S3
+	}
+
 	clientOperationID, err := clientConn.ExportToS3(ctx, client, s3Settings, featureFlags)
 	if err != nil {
 		xlog.Error(ctx, "can't start export operation", zap.Error(err))
@@ -379,9 +422,10 @@ func MakeBackup(
 			CreatedAt: now,
 			Creator:   subject,
 		},
-		ScheduleID:  req.ScheduleID,
-		ExpireAt:    expireAt,
-		SourcePaths: pathsForExport,
+		ScheduleID:         req.ScheduleID,
+		ExpireAt:           expireAt,
+		SourcePaths:        pathsForExport,
+		EncryptionSettings: req.EncryptionSettings,
 	}
 
 	op := &types.TakeBackupOperation{
@@ -399,9 +443,10 @@ func MakeBackup(
 			CreatedAt: now,
 			Creator:   subject,
 		},
-		YdbOperationId:    clientOperationID,
-		UpdatedAt:         now,
-		ParentOperationID: req.ParentOperationID,
+		YdbOperationId:     clientOperationID,
+		UpdatedAt:          now,
+		ParentOperationID:  req.ParentOperationID,
+		EncryptionSettings: req.EncryptionSettings,
 	}
 
 	return backup, op, nil

internal/config/config.go

@@ -66,8 +66,9 @@ type MetricsServerConfig struct {
 }
 
 type FeatureFlagsConfig struct {
-	DisableTTLDeletion   bool `yaml:"disable_ttl_deletion" default:"false"`
-	EnableNewPathsFormat bool `yaml:"enable_new_paths_format" default:"false"`
+	DisableTTLDeletion     bool `yaml:"disable_ttl_deletion" default:"false"`
+	EnableNewPathsFormat   bool `yaml:"enable_new_paths_format" default:"false"`
+	EnableBackupEncryption bool `yaml:"enable_backup_encryption" default:"false"`
 }
 
 type LogConfig struct {

internal/connectors/client/connector.go

@@ -272,6 +272,17 @@ func (d *ClientYdbConnector) ExportToS3(
 		exportRequest.Settings.DestinationPrefix = s3Settings.DestinationPrefix
 	}
 
+	if featureFlags.EnableBackupEncryption && len(s3Settings.EncryptionKey) > 0 {
+		exportRequest.Settings.EncryptionSettings = &Ydb_Export.EncryptionSettings{
+			EncryptionAlgorithm: s3Settings.EncryptionAlgorithm,
+			Key: &Ydb_Export.EncryptionSettings_SymmetricKey_{
+				SymmetricKey: &Ydb_Export.EncryptionSettings_SymmetricKey{
+					Key: s3Settings.EncryptionKey,
+				},
+			},
+		}
+	}
+
 	response, err := exportClient.ExportToS3(ctx, exportRequest)
 
 	if err != nil {
@@ -425,6 +436,17 @@ func (d *ClientYdbConnector) ImportFromS3(
 		importRequest.Settings.DestinationPath = path.Join(clientDb.Name(), s3Settings.DestinationPrefix)
 	}
 
+	if len(s3Settings.EncryptionKey) > 0 {
+		importRequest.Settings.EncryptionSettings = &Ydb_Export.EncryptionSettings{
+			EncryptionAlgorithm: s3Settings.EncryptionAlgorithm,
+			Key: &Ydb_Export.EncryptionSettings_SymmetricKey_{
+				SymmetricKey: &Ydb_Export.EncryptionSettings_SymmetricKey{
+					Key: s3Settings.EncryptionKey,
+				},
+			},
+		}
+	}
+
 	response, err := importClient.ImportFromS3(ctx, importRequest)
 
 	if err != nil {

internal/connectors/db/yql/schema/create_tables.yql

@@ -22,6 +22,9 @@ CREATE TABLE Backups (
 
     schedule_id String,
 
+    encryption_algorithm String,
+    kms_key_id String,
+
     INDEX idx_container_id GLOBAL ON (container_id),
     INDEX idx_created_at GLOBAL ON (created_at),
     INDEX idx_expire_at GLOBAL ON (status, expire_at),
@@ -55,6 +58,8 @@ CREATE TABLE Operations (
     paths_to_exclude String,
     operation_id String,
     parent_operation_id String,
+    encryption_algorithm String,
+    kms_key_id String,
     --used only in TBWR
     schedule_id String,
     ttl Interval,
@@ -85,6 +90,9 @@ CREATE TABLE BackupSchedules (
     initiated String,
     created_at Timestamp,
 
+    encryption_algorithm String,
+    kms_key_id String,
+
     recovery_point_objective Interval,
 
     next_launch Timestamp,

internal/handlers/schedule_backup.go

@@ -8,6 +8,7 @@ import (
 	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 	"ydbcp/internal/audit"
+	"ydbcp/internal/config"
 	"ydbcp/internal/connectors/db"
 	"ydbcp/internal/connectors/db/yql/queries"
 	"ydbcp/internal/types"
@@ -20,11 +21,13 @@ type BackupScheduleHandlerType func(context.Context, db.DBConnector, *types.Back
 func NewBackupScheduleHandler(
 	queryBuilderFactory queries.WriteQueryBuilderFactory,
 	clock clockwork.Clock,
+	featureFlags config.FeatureFlagsConfig,
 ) BackupScheduleHandlerType {
 	return func(ctx context.Context, driver db.DBConnector, schedule *types.BackupSchedule) error {
 		return BackupScheduleHandler(
 			ctx, driver, schedule,
 			queryBuilderFactory, clock,
+			featureFlags,
 		)
 	}
 }
@@ -46,6 +49,7 @@ func BackupScheduleHandler(
 	schedule *types.BackupSchedule,
 	queryBuilderFactory queries.WriteQueryBuilderFactory,
 	clock clockwork.Clock,
+	featureFlags config.FeatureFlagsConfig,
 ) error {
 	if schedule.Status != types.BackupScheduleStateActive {
 		xlog.Error(ctx, "backup schedule is not active", zap.String("ScheduleID", schedule.ID))
@@ -85,6 +89,10 @@ func BackupScheduleHandler(
 				d := schedule.ScheduleSettings.Ttl.AsDuration()
 				tbwr.Ttl = &d
 			}
+
+			if schedule.ScheduleSettings.EncryptionSettings != nil && featureFlags.EnableBackupEncryption {
+				tbwr.EncryptionSettings = schedule.ScheduleSettings.EncryptionSettings
+			}
 		}
 
 		xlog.Info(

internal/handlers/schedule_backup_test.go

@@ -6,6 +6,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"testing"
 	"time"
+	"ydbcp/internal/config"
 	"ydbcp/internal/connectors/db"
 	"ydbcp/internal/connectors/db/yql/queries"
 	"ydbcp/internal/types"
@@ -40,7 +41,7 @@ func TestBackupScheduleHandler(t *testing.T) {
 	)
 
 	handler := NewBackupScheduleHandler(
-		queries.NewWriteTableQueryMock, clock,
+		queries.NewWriteTableQueryMock, clock, config.FeatureFlagsConfig{},
 	)
 	err := handler(ctx, dbConnector, &schedule)
 	assert.Empty(t, err)

internal/server/services/backup/backup_service.go

@@ -3,6 +3,7 @@ package backup
 import (
 	"context"
 	"github.com/jonboulle/clockwork"
+	"google.golang.org/protobuf/proto"
 	"strconv"
 	"time"
 	"ydbcp/internal/audit"
@@ -110,9 +111,13 @@ func (s *BackupService) MakeBackup(ctx context.Context, req *pb.MakeBackupReques
 	ctx = xlog.With(ctx, zap.String("SubjectID", subject))
 	now := timestamppb.Now()
 
+	var encryptionSettings *pb.EncryptionSettings
 	if req.EncryptionSettings != nil {
-		s.IncApiCallsCounter(methodName, codes.Unimplemented)
-		return nil, status.Error(codes.Unimplemented, "backup encryption is not supported yet")
+		if !s.featureFlags.EnableBackupEncryption {
+			s.IncApiCallsCounter(methodName, codes.Unimplemented)
+			return nil, status.Error(codes.Unimplemented, "backup encryption is not supported yet")
+		}
+		encryptionSettings = proto.Clone(req.EncryptionSettings).(*pb.EncryptionSettings)
 	}
 
 	tbwr := &types.TakeBackupWithRetryOperation{
@@ -130,7 +135,8 @@ func (s *BackupService) MakeBackup(ctx context.Context, req *pb.MakeBackupReques
 				Creator:   subject,
 				CreatedAt: now,
 			},
-			UpdatedAt: now,
+			UpdatedAt:          now,
+			EncryptionSettings: encryptionSettings,
 		},
 		Retries: 0,
 		RetryConfig: &pb.RetryConfig{
@@ -383,6 +389,14 @@ func (s *BackupService) MakeRestore(ctx context.Context, req *pb.MakeRestoreRequ
 		DestinationPrefix: req.GetDestinationPrefix(),
 	}
 
+	if backup.EncryptionSettings != nil {
+		// TODO: read encrypted DEK from s3
+		// TODO: decrypt DEC via KMS
+		// TODO: set EncryptionKey
+		_, algorithm, _ := backup_operations.GetEncryptionParams(backup.EncryptionSettings)
+		s3Settings.EncryptionAlgorithm = algorithm
+	}
+
 	clientOperationID, err := s.clientConn.ImportFromS3(ctx, clientDriver, s3Settings, s.featureFlags)
 	if err != nil {
 		xlog.Error(ctx, "can't start import operation", zap.Error(err))

internal/server/services/backup_schedule/backup_schedule_service.go

@@ -119,7 +119,7 @@ func (s *BackupScheduleService) CreateBackupSchedule(
 		return nil, status.Error(codes.FailedPrecondition, "no backup schedule settings for CreateBackupSchedule")
 	}
 
-	if request.ScheduleSettings.EncryptionSettings != nil {
+	if !s.config.FeatureFlags.EnableBackupEncryption && request.ScheduleSettings.EncryptionSettings != nil {
 		s.IncApiCallsCounter(methodName, codes.Unimplemented)
 		return nil, status.Error(codes.Unimplemented, "backup encryption is not supported yet")
 	}

internal/types/backup.go

@@ -37,21 +37,22 @@ func ParseObjectID(string string) (string, error) {
 }
 
 type Backup struct {
-	ID               string
-	ContainerID      string
-	DatabaseName     string
-	DatabaseEndpoint string
-	S3Endpoint       string
-	S3Region         string
-	S3Bucket         string
-	S3PathPrefix     string
-	Status           string
-	Message          string
-	AuditInfo        *pb.AuditInfo
-	Size             int64
-	ScheduleID       *string
-	ExpireAt         *time.Time
-	SourcePaths      []string
+	ID                 string
+	ContainerID        string
+	DatabaseName       string
+	DatabaseEndpoint   string
+	S3Endpoint         string
+	S3Region           string
+	S3Bucket           string
+	S3PathPrefix       string
+	Status             string
+	Message            string
+	AuditInfo          *pb.AuditInfo
+	Size               int64
+	ScheduleID         *string
+	ExpireAt           *time.Time
+	SourcePaths        []string
+	EncryptionSettings *pb.EncryptionSettings
 }
 
 func (o *Backup) String() string {
@@ -77,12 +78,13 @@ func (o *Backup) Proto() *pb.Backup {
 			Bucket:     o.S3Bucket,
 			PathPrefix: o.S3PathPrefix,
 		},
-		Audit:       o.AuditInfo,
-		Size:        o.Size,
-		Status:      pb.Backup_Status(pb.Backup_Status_value[o.Status]),
-		Message:     o.Message,
-		ExpireAt:    nil,
-		SourcePaths: o.SourcePaths,
+		Audit:              o.AuditInfo,
+		Size:               o.Size,
+		Status:             pb.Backup_Status(pb.Backup_Status_value[o.Status]),
+		Message:            o.Message,
+		ExpireAt:           nil,
+		SourcePaths:        o.SourcePaths,
+		EncryptionSettings: o.EncryptionSettings,
 	}
 	if o.ScheduleID != nil {
 		backup.ScheduleId = *o.ScheduleID