feat: add GCS customer-supplied encryption key (CSEK) support

Add support for encrypting GCS backup data using customer-supplied
encryption keys (CSEK). This provides client-side encryption where
the encryption key is controlled by the user, not Google.

Changes:
- Add `encryption_key` config field to GCSConfig (GCS_ENCRYPTION_KEY env var)
- Validate and decode base64 encryption key on connect (must be 256-bit)
- Apply encryption to all object operations: read, write, stat, copy
- Update documentation with usage instructions

Usage:
  gcs:
    encryption_key: ""  # base64-encoded 256-bit key
                        # Generate with: openssl rand -base64 32

See: https://cloud.google.com/storage/docs/encryption/customer-supplied-keys

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: ns-janandaram

ReadMe.md

@@ -304,6 +304,10 @@ gcs:
   custom_storage_class_map: {}
   debug: false                 # GCS_DEBUG
   force_http: false            # GCS_FORCE_HTTP
+  # GCS_ENCRYPTION_KEY, base64-encoded 256-bit key for customer-supplied encryption (CSEK)
+  # This encrypts backup data at rest using a key you control. Generate with: `openssl rand -base64 32`
+  # See https://cloud.google.com/storage/docs/encryption/customer-supplied-keys
+  encryption_key: ""
 cos:
   url: ""                      # COS_URL
   timeout: 2m                  # COS_TIMEOUT

pkg/config/config.go

@@ -113,6 +113,9 @@ type GCSConfig struct {
 	// 			UploadConcurrency or DownloadConcurrency in each upload and download case
 	ClientPoolSize int `yaml:"client_pool_size" envconfig:"GCS_CLIENT_POOL_SIZE"`
 	ChunkSize      int `yaml:"chunk_size" envconfig:"GCS_CHUNK_SIZE"`
+	// EncryptionKey is a base64-encoded 256-bit customer-supplied encryption key (CSEK)
+	// for client-side encryption of objects. Use `openssl rand -base64 32` to generate.
+	EncryptionKey string `yaml:"encryption_key" envconfig:"GCS_ENCRYPTION_KEY"`
 }
 
 // AzureBlobConfig - Azure Blob settings section

pkg/storage/gcs.go

@@ -26,9 +26,10 @@ import (
 
 // GCS - presents methods for manipulate data on GCS
 type GCS struct {
-	client     *storage.Client
-	Config     *config.GCSConfig
-	clientPool *pool.ObjectPool
+	client        *storage.Client
+	Config        *config.GCSConfig
+	clientPool    *pool.ObjectPool
+	encryptionKey []byte // Customer-Supplied Encryption Key (CSEK)
 }
 
 type debugGCSTransport struct {
@@ -188,14 +189,39 @@ func (gcs *GCS) Connect(ctx context.Context) error {
 	gcs.clientPool = pool.NewObjectPoolWithDefaultConfig(ctx, factory)
 	gcs.clientPool.Config.MaxTotal = gcs.Config.ClientPoolSize * 3
 	gcs.client, err = storage.NewClient(ctx, storageClientOptions...)
-	return err
+	if err != nil {
+		return err
+	}
+
+	// Validate and decode the encryption key if provided
+	if gcs.Config.EncryptionKey != "" {
+		key, err := base64.StdEncoding.DecodeString(gcs.Config.EncryptionKey)
+		if err != nil {
+			return errors.Wrap(err, "gcs: malformed encryption_key, must be base64-encoded 256-bit key")
+		}
+		if len(key) != 32 {
+			return fmt.Errorf("gcs: malformed encryption_key, must be base64-encoded 256-bit key (got %d bytes)", len(key))
+		}
+		gcs.encryptionKey = key
+		log.Info().Msg("GCS: Customer-Supplied Encryption Key (CSEK) configured")
+	}
+
+	return nil
 }
 
 func (gcs *GCS) Close(ctx context.Context) error {
 	gcs.clientPool.Close(ctx)
 	return gcs.client.Close()
 }
 
+// applyEncryption returns an ObjectHandle with encryption key applied if configured
+func (gcs *GCS) applyEncryption(obj *storage.ObjectHandle) *storage.ObjectHandle {
+	if gcs.encryptionKey != nil {
+		return obj.Key(gcs.encryptionKey)
+	}
+	return obj
+}
+
 func (gcs *GCS) Walk(ctx context.Context, gcsPath string, recursive bool, process func(ctx context.Context, r RemoteFile) error) error {
 	rootPath := path.Join(gcs.Config.Path, gcsPath)
 	return gcs.WalkAbsolute(ctx, rootPath, recursive, process)
@@ -252,7 +278,7 @@ func (gcs *GCS) GetFileReaderAbsolute(ctx context.Context, key string) (io.ReadC
 		return nil, err
 	}
 	pClient := pClientObj.(*clientObject).Client
-	obj := pClient.Bucket(gcs.Config.Bucket).Object(key)
+	obj := gcs.applyEncryption(pClient.Bucket(gcs.Config.Bucket).Object(key))
 	reader, err := obj.NewReader(ctx)
 	if err != nil {
 		if pErr := gcs.clientPool.InvalidateObject(ctx, pClientObj); pErr != nil {
@@ -281,7 +307,7 @@ func (gcs *GCS) PutFileAbsolute(ctx context.Context, key string, r io.ReadCloser
 		return err
 	}
 	pClient := pClientObj.(*clientObject).Client
-	obj := pClient.Bucket(gcs.Config.Bucket).Object(key)
+	obj := gcs.applyEncryption(pClient.Bucket(gcs.Config.Bucket).Object(key))
 	// always retry transient errors to mitigate retry logic bugs.
 	obj = obj.Retryer(storage.WithPolicy(storage.RetryAlways))
 	writer := obj.NewWriter(ctx)
@@ -314,7 +340,8 @@ func (gcs *GCS) StatFile(ctx context.Context, key string) (RemoteFile, error) {
 }
 
 func (gcs *GCS) StatFileAbsolute(ctx context.Context, key string) (RemoteFile, error) {
-	objAttr, err := gcs.client.Bucket(gcs.Config.Bucket).Object(key).Attrs(ctx)
+	obj := gcs.applyEncryption(gcs.client.Bucket(gcs.Config.Bucket).Object(key))
+	objAttr, err := obj.Attrs(ctx)
 	if err != nil {
 		if errors.Is(err, storage.ErrObjectNotExist) {
 			return nil, ErrNotFound
@@ -369,7 +396,7 @@ func (gcs *GCS) CopyObject(ctx context.Context, srcSize int64, srcBucket, srcKey
 	}
 	pClient := pClientObj.(*clientObject).Client
 	src := pClient.Bucket(srcBucket).Object(srcKey)
-	dst := pClient.Bucket(gcs.Config.Bucket).Object(dstKey)
+	dst := gcs.applyEncryption(pClient.Bucket(gcs.Config.Bucket).Object(dstKey))
 	// always retry transient errors to mitigate retry logic bugs.
 	dst = dst.Retryer(storage.WithPolicy(storage.RetryAlways))
 	attrs, err := src.Attrs(ctx)
@@ -379,7 +406,10 @@ func (gcs *GCS) CopyObject(ctx context.Context, srcSize int64, srcBucket, srcKey
 		}
 		return 0, err
 	}
-	if _, err = dst.CopierFrom(src).Run(ctx); err != nil {
+	copier := dst.CopierFrom(src)
+	// If encryption is enabled, the destination will be encrypted
+	// Note: source objects from object disks are not encrypted by clickhouse-backup
+	if _, err = copier.Run(ctx); err != nil {
 		if pErr := gcs.clientPool.InvalidateObject(ctx, pClientObj); pErr != nil {
 			log.Warn().Msgf("gcs.CopyObject: gcs.clientPool.InvalidateObject error: %+v", pErr)
 		}