feat(ydbcp): implement kms plugin

Author: ulya-sidorina

internal/auth/auth.go

@@ -27,7 +27,7 @@ var (
 	ErrGetAuthToken = errors.New("can't get auth token")
 )
 
-func NewAuthProvider(ctx context.Context, cfg config.AuthConfig) (auth.AuthProvider, error) {
+func NewAuthProvider(ctx context.Context, cfg config.PluginConfig) (auth.AuthProvider, error) {
 	xlog.Info(ctx, "Loading auth provider plugin", zap.String("PluginPath", cfg.PluginPath))
 
 	plug, err := plugin.Open(cfg.PluginPath)

internal/config/config.go

@@ -45,7 +45,7 @@ type ClientConnectionConfig struct {
 	AllowInsecureEndpoint  bool     `yaml:"allow_insecure_endpoint" default:"false"`
 }
 
-type AuthConfig struct {
+type PluginConfig struct {
 	PluginPath    string      `yaml:"plugin_path"`
 	Configuration interface{} `yaml:"configuration"`
 }
@@ -95,7 +95,7 @@ type Config struct {
 	DBConnection       YDBConnectionConfig      `yaml:"db_connection"`
 	ClientConnection   ClientConnectionConfig   `yaml:"client_connection"`
 	S3                 S3Config                 `yaml:"s3"`
-	Auth               AuthConfig               `yaml:"auth"`
+	Auth               PluginConfig             `yaml:"auth"`
 	GRPCServer         GRPCServerConfig         `yaml:"grpc_server"`
 	MetricsServer      MetricsServerConfig      `yaml:"metrics_server"`
 	OperationProcessor OperationProcessorConfig `yaml:"operation_processor"`
@@ -223,7 +223,7 @@ func (c *S3Config) SecretKey() (string, error) {
 
 }
 
-func (c *AuthConfig) ConfigurationString() (string, error) {
+func (c *PluginConfig) ConfigurationString() (string, error) {
 	txt, err := yaml.Marshal(c.Configuration)
 	if err != nil {
 		return "", fmt.Errorf("can't marshal Auth.Configuration to YAML: %w", err)

internal/kms/kms.go

@@ -0,0 +1,39 @@
+package kms
+
+import (
+	"context"
+	"fmt"
+	"plugin"
+	"ydbcp/internal/config"
+	"ydbcp/internal/util/xlog"
+	"ydbcp/pkg/plugins/kms"
+
+	"go.uber.org/zap"
+)
+
+func NewKmsProvider(ctx context.Context, cfg config.PluginConfig) (kms.KmsProvider, error) {
+	xlog.Info(ctx, "Loading kms provider plugin", zap.String("PluginPath", cfg.PluginPath))
+
+	pluginInstance, err := plugin.Open(cfg.PluginPath)
+	if err != nil {
+		return nil, fmt.Errorf("can't load kms provider plugin, path %s: %w", cfg.PluginPath, err)
+	}
+	symbol, err := pluginInstance.Lookup("KmsProvider")
+	if err != nil {
+		return nil, fmt.Errorf("can't lookup KmsProvider symbol, plugin path %s: %w", cfg.PluginPath, err)
+	}
+
+	var instance kms.KmsProvider
+	instance, ok := symbol.(kms.KmsProvider)
+	if !ok {
+		return nil, fmt.Errorf("can't cast KmsProvider symbol, plugin path %s", cfg.PluginPath)
+	}
+	pluginConfig, err := cfg.ConfigurationString()
+	if err != nil {
+		return nil, fmt.Errorf("can't get kms provider configuration: %w", err)
+	}
+	if err = instance.Init(ctx, pluginConfig); err != nil {
+		return nil, fmt.Errorf("can't initialize kms provider plugin: %w", err)
+	}
+	return instance, nil
+}

internal/kms/mock.go

@@ -0,0 +1,3 @@
+
+
+// TODO: Implement

internal/util/tls_setup/tls_setup.go

@@ -5,17 +5,53 @@ import (
 	"crypto/x509"
 	"errors"
 	"fmt"
+	"os"
+
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
-	"os"
 )
 
-func LoadTLSCredentials(RootCAPath *string, withInsecure bool) (grpc.DialOption, error) {
+func LoadTLSCredentials(rootCAPath *string, withInsecure bool) (grpc.DialOption, error) {
+	tlsConfig, err := buildTLSConfig(rootCAPath, withInsecure)
+	if err != nil {
+		return nil, err
+	}
+	return grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)), nil
+}
+
+func LoadMTLSCredentials(
+	rootCAPath *string,
+	clientCertPath *string,
+	clientKeyPath *string,
+	withInsecure bool,
+) (grpc.DialOption, error) {
+	if clientCertPath == nil || len(*clientCertPath) == 0 {
+		return nil, fmt.Errorf("client certificate path is required for mTLS")
+	}
+	if clientKeyPath == nil || len(*clientKeyPath) == 0 {
+		return nil, fmt.Errorf("client key path is required for mTLS")
+	}
+
+	tlsConfig, err := buildTLSConfig(rootCAPath, withInsecure)
+	if err != nil {
+		return nil, err
+	}
+
+	cert, err := tls.LoadX509KeyPair(*clientCertPath, *clientKeyPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load client certificate/key: %w", err)
+	}
+	tlsConfig.Certificates = []tls.Certificate{cert}
+
+	return grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)), nil
+}
+
+func buildTLSConfig(rootCAPath *string, withInsecure bool) (*tls.Config, error) {
 	var certPool *x509.CertPool
-	if RootCAPath != nil && len(*RootCAPath) > 0 {
-		caBundle, err := os.ReadFile(*RootCAPath)
+	if rootCAPath != nil && len(*rootCAPath) > 0 {
+		caBundle, err := os.ReadFile(*rootCAPath)
 		if err != nil {
-			return nil, fmt.Errorf("unable to read root ca bundle from file %s: %w", *RootCAPath, err)
+			return nil, fmt.Errorf("unable to read root ca bundle from file %s: %w", *rootCAPath, err)
 		}
 		certPool = x509.NewCertPool()
 		if ok := certPool.AppendCertsFromPEM(caBundle); !ok {
@@ -36,5 +72,5 @@ func LoadTLSCredentials(RootCAPath *string, withInsecure bool) (grpc.DialOption,
 	if withInsecure {
 		tlsConfig.InsecureSkipVerify = true
 	}
-	return grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)), nil
+	return tlsConfig, nil
 }

pkg/plugins/kms/kms.go

@@ -0,0 +1,32 @@
+package kms
+
+import "context"
+
+type EncryptRequest struct {
+	KeyID     string
+	Plaintext []byte
+}
+
+type EncryptResponse struct {
+	KeyID      string
+	Ciphertext []byte
+}
+
+type DecryptRequest struct {
+	KeyID      string
+	Ciphertext []byte
+}
+
+type DecryptResponse struct {
+	KeyID     string
+	Plaintext []byte
+}
+
+// KmsProvider is an interface that KMS plugins must implement.
+type KmsProvider interface {
+	Init(ctx context.Context, config string) error
+	Close(ctx context.Context) error
+
+	Encrypt(ctx context.Context, req *EncryptRequest) (*EncryptResponse, error)
+	Decrypt(ctx context.Context, req *DecryptRequest) (*DecryptResponse, error)
+}

plugins/kms_nebius/kms_nebius.go

@@ -0,0 +1,127 @@
+package main
+
+import (
+	"context"
+	"fmt"
+
+	"ydbcp/internal/util/tls_setup"
+	"ydbcp/internal/util/xlog"
+	"ydbcp/pkg/plugins/kms"
+	pb "ydbcp/plugins/kms_nebius/proto"
+
+	"go.uber.org/zap"
+	"google.golang.org/grpc"
+	"gopkg.in/yaml.v3"
+)
+
+type kmsProviderNebius struct {
+	config pluginConfig
+}
+
+type pluginConfig struct {
+	CryptoServiceEndpoint string `yaml:"crypto_service_endpoint"`
+	Insecure              bool   `yaml:"insecure" default:"false"`
+	RootCAPath            string `yaml:"root_ca_path"`
+	ClientKeyPath         string `yaml:"client_key_path"`
+	ClientCertificatePath string `yaml:"client_certificate_path"`
+}
+
+func (p *kmsProviderNebius) Init(ctx context.Context, rawConfig string) error {
+	xlog.Info(ctx, "KmsNebiusProvider initialization started", zap.String("config", rawConfig))
+	if err := yaml.Unmarshal([]byte(rawConfig), &p.config); err != nil {
+		xlog.Error(ctx, "Unable to parse configuration", zap.Error(err))
+		return fmt.Errorf("kms: unable to parse configuration: %w", err)
+	}
+	if len(p.config.CryptoServiceEndpoint) == 0 {
+		return fmt.Errorf("kms: crypto service endpoint is required in configuration")
+	}
+	if len(p.config.RootCAPath) == 0 {
+		return fmt.Errorf("kms: root ca path is required in configuration")
+	}
+	if len(p.config.ClientKeyPath) == 0 {
+		return fmt.Errorf("kms: client key path is required in configuration")
+	}
+	if len(p.config.ClientCertificatePath) == 0 {
+		return fmt.Errorf("kms: client certificate path is required in configuration")
+	}
+
+	xlog.Info(ctx, "KmsNebiusProvider was initialized successfully")
+	return nil
+}
+
+func (p *kmsProviderNebius) Close(ctx context.Context) error {
+	xlog.Info(ctx, "KmsNebiusProvider was closed")
+	return nil
+}
+
+func (p *kmsProviderNebius) Encrypt(ctx context.Context, req *kms.EncryptRequest) (*kms.EncryptResponse, error) {
+	if req == nil {
+		return nil, fmt.Errorf("kms: encryption request is nil")
+	}
+	if len(req.KeyID) == 0 {
+		return nil, fmt.Errorf("kms: key id is required in encryption request")
+	}
+
+	tlsOption, err := tls_setup.LoadMTLSCredentials(&p.config.RootCAPath, &p.config.ClientCertificatePath, &p.config.ClientKeyPath, p.config.Insecure)
+	if err != nil {
+		return nil, err
+	}
+
+	grpcClient, err := grpc.NewClient("dns:"+p.config.CryptoServiceEndpoint, tlsOption) // TODO: do we need dns prefix?
+	if err != nil {
+		return nil, err
+	}
+	defer grpcClient.Close()
+
+	client := pb.NewSymmetricCryptoServiceClient(grpcClient)
+	resp, err := client.Encrypt(ctx, &pb.SymmetricEncryptRequest{
+		KeyId:     req.KeyID,
+		Plaintext: req.Plaintext,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("kms: encryption was failed: %w", err)
+	}
+
+	return &kms.EncryptResponse{
+		KeyID:      resp.GetKeyId(),
+		Ciphertext: resp.GetCiphertext(),
+	}, nil
+}
+
+func (p *kmsProviderNebius) Decrypt(ctx context.Context, req *kms.DecryptRequest) (*kms.DecryptResponse, error) {
+	if req == nil {
+		return nil, fmt.Errorf("kms: decryption request is nil")
+	}
+	if len(req.KeyID) == 0 {
+		return nil, fmt.Errorf("kms: key id is required in decryption request")
+	}
+
+	tlsOption, err := tls_setup.LoadTLSCredentials(&p.config.RootCAPath, p.config.Insecure)
+	if err != nil {
+		return nil, err
+	}
+
+	grpcClient, err := grpc.NewClient("dns:"+p.config.CryptoServiceEndpoint, tlsOption) // TODO: do we need dns prefix?
+	if err != nil {
+		return nil, err
+	}
+	defer grpcClient.Close()
+
+	client := pb.NewSymmetricCryptoServiceClient(grpcClient)
+	resp, err := client.Decrypt(ctx, &pb.SymmetricDecryptRequest{
+		KeyId:      req.KeyID,
+		Ciphertext: req.Ciphertext,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("kms: decryption was failed: %w", err)
+	}
+
+	return &kms.DecryptResponse{
+		KeyID:     resp.GetKeyId(),
+		Plaintext: resp.GetPlaintext(),
+	}, nil
+}
+
+func main() {}
+
+var KmsProvider kmsProviderNebius