kmsv2: add grpc service

Signed-off-by: Krzysztof Ostrowski <kostrows@redhat.com>

staging/src/k8s.io/kms/go.mod

@@ -7,9 +7,11 @@ go 1.19
 require (
 	github.com/gogo/protobuf v1.3.2
 	google.golang.org/grpc v1.51.0
+	k8s.io/klog/v2 v2.80.1
 )
 
 require (
+	github.com/go-logr/logr v1.2.3 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/google/go-cmp v0.5.9 // indirect
 	golang.org/x/net v0.4.0 // indirect

staging/src/k8s.io/kms/go.sum

@@ -19,6 +19,9 @@ github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.m
 github.com/envoyproxy/go-control-plane v0.10.2-0.20220325020618-49ff273808a1/go.mod h1:KJwIaB5Mv44NWtYuAOFCVOjcI94vtpEz2JU/D2v6IjE=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
+github.com/go-logr/logr v1.2.0/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
+github.com/go-logr/logr v1.2.3 h1:2DntVwHkVopvECVRSlL5PSo9eG+cAkDCuckLubN+rq0=
+github.com/go-logr/logr v1.2.3/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
@@ -156,3 +159,5 @@ gopkg.in/yaml.v2 v2.2.3/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
+k8s.io/klog/v2 v2.80.1 h1:atnLQ121W371wYYFawwYx1aEY2eUfs4l3J72wtgAwV4=
+k8s.io/klog/v2 v2.80.1/go.mod h1:y1WjHnz7Dj687irZUWR/WLkLc5N1YHtjLdmgWjndZn0=

staging/src/k8s.io/kms/service/grpc_service.go

@@ -0,0 +1,141 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package service
+
+import (
+	"context"
+	"net"
+	"time"
+
+	"google.golang.org/grpc"
+
+	"k8s.io/klog/v2"
+	kmsapi "k8s.io/kms/apis/v2alpha1"
+)
+
+// GRPCService is a gprc server that runs the kms v2 alpha 1 API.
+type GRPCService struct {
+	addr    string
+	timeout time.Duration
+	server  *grpc.Server
+
+	kmsService Service
+}
+
+var _ kmsapi.KeyManagementServiceServer = (*GRPCService)(nil)
+
+// NewGRPCService creates an instance of GRPCService.
+func NewGRPCService(
+	address string,
+	timeout time.Duration,
+
+	kmsService Service,
+) *GRPCService {
+	klog.V(4).InfoS("KMS plugin configured", "address", address, "timeout", timeout)
+
+	return &GRPCService{
+		addr:       address,
+		timeout:    timeout,
+		kmsService: kmsService,
+	}
+}
+
+// ListenAndServe accepts incoming connections on a Unix socket. It is a blocking method.
+// Returns non-nil error unless Close or Shutdown is called.
+func (s *GRPCService) ListenAndServe() error {
+	ln, err := net.Listen("unix", s.addr)
+	if err != nil {
+		return err
+	}
+	defer ln.Close()
+
+	gs := grpc.NewServer(
+		grpc.ConnectionTimeout(s.timeout),
+	)
+	s.server = gs
+
+	kmsapi.RegisterKeyManagementServiceServer(gs, s)
+
+	klog.V(4).InfoS("kms plugin serving", "address", s.addr)
+	return gs.Serve(ln)
+}
+
+// Shutdown performs a grafecul shutdown. Doesn't accept new connections and
+// blocks until all pending  RPCs are finished.
+func (s *GRPCService) Shutdown() {
+	klog.V(4).InfoS("kms plugin shutdown", "address", s.addr)
+	if s.server != nil {
+		s.server.GracefulStop()
+	}
+}
+
+// Close stops the server by closing all connections immediately and cancels
+// all active RPCs.
+func (s *GRPCService) Close() {
+	klog.V(4).InfoS("kms plugin close", "address", s.addr)
+	if s.server != nil {
+		s.server.Stop()
+	}
+}
+
+// Status sends a status request to specified kms service.
+func (s *GRPCService) Status(ctx context.Context, _ *kmsapi.StatusRequest) (*kmsapi.StatusResponse, error) {
+	res, err := s.kmsService.Status(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return &kmsapi.StatusResponse{
+		Version: res.Version,
+		Healthz: res.Healthz,
+		KeyId:   res.KeyID,
+	}, nil
+}
+
+// Decrypt sends a decryption request to specified kms service.
+func (s *GRPCService) Decrypt(ctx context.Context, req *kmsapi.DecryptRequest) (*kmsapi.DecryptResponse, error) {
+	klog.V(4).InfoS("decrypt request received", "id", req.Uid)
+
+	plaintext, err := s.kmsService.Decrypt(ctx, req.Uid, &DecryptRequest{
+		Ciphertext:  req.Ciphertext,
+		KeyID:       req.KeyId,
+		Annotations: req.Annotations,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &kmsapi.DecryptResponse{
+		Plaintext: plaintext,
+	}, nil
+}
+
+// Encrypt sends an encryption request to specified kms service.
+func (s *GRPCService) Encrypt(ctx context.Context, req *kmsapi.EncryptRequest) (*kmsapi.EncryptResponse, error) {
+	klog.V(4).InfoS("encrypt request received", "id", req.Uid)
+
+	encRes, err := s.kmsService.Encrypt(ctx, req.Uid, req.Plaintext)
+	if err != nil {
+		return nil, err
+	}
+
+	return &kmsapi.EncryptResponse{
+		Ciphertext:  encRes.Ciphertext,
+		KeyId:       encRes.KeyID,
+		Annotations: encRes.Annotations,
+	}, nil
+}

staging/src/k8s.io/kms/service/grpc_service_test.go

@@ -0,0 +1,192 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package service
+
+import (
+	"bytes"
+	"context"
+	"encoding/base64"
+	"fmt"
+	"math/rand"
+	"net"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	kmsapi "k8s.io/kms/apis/v2alpha1"
+)
+
+const version = "v2alpha1"
+
+func TestGRPCService(t *testing.T) {
+	t.Parallel()
+
+	defaultTimeout := 30 * time.Second
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
+	t.Cleanup(cancel)
+
+	address := filepath.Join(os.TempDir(), "kmsv2.sock")
+	plaintext := []byte("lorem ipsum dolor sit amet")
+	r := rand.New(rand.NewSource(time.Now().Unix()))
+	id, err := makeID(r.Read)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	kmsService := newBase64Service(id)
+	server := NewGRPCService(address, defaultTimeout, kmsService)
+	go func() {
+		if err := server.ListenAndServe(); err != nil {
+			panic(err)
+		}
+	}()
+	t.Cleanup(server.Shutdown)
+
+	client := newClient(t, address)
+
+	t.Run("should be able to encrypt and decrypt through unix domain sockets", func(t *testing.T) {
+		t.Parallel()
+
+		encRes, err := client.Encrypt(ctx, &kmsapi.EncryptRequest{
+			Plaintext: plaintext,
+			Uid:       id,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if bytes.Equal(plaintext, encRes.Ciphertext) {
+			t.Fatal("plaintext and ciphertext shouldn't be equal!")
+		}
+
+		decRes, err := client.Decrypt(ctx, &kmsapi.DecryptRequest{
+			Ciphertext:  encRes.Ciphertext,
+			KeyId:       encRes.KeyId,
+			Annotations: encRes.Annotations,
+			Uid:         id,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if !bytes.Equal(decRes.Plaintext, plaintext) {
+			t.Errorf("want: %q, have: %q", plaintext, decRes.Plaintext)
+		}
+	})
+
+	t.Run("should return status data", func(t *testing.T) {
+		t.Parallel()
+
+		status, err := client.Status(ctx, &kmsapi.StatusRequest{})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if status.Healthz != "ok" {
+			t.Errorf("want: %q, have: %q", "ok", status.Healthz)
+		}
+		if len(status.KeyId) == 0 {
+			t.Errorf("want: len(keyID) > 0, have: %d", len(status.KeyId))
+		}
+		if status.Version != version {
+			t.Errorf("want %q, have: %q", version, status.Version)
+		}
+	})
+}
+
+func newClient(t *testing.T, address string) kmsapi.KeyManagementServiceClient {
+	t.Helper()
+
+	cnn, err := grpc.Dial(
+		address,
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithDialer(func(addr string, t time.Duration) (net.Conn, error) {
+			return net.Dial("unix", addr)
+		}),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	t.Cleanup(func() { _ = cnn.Close() })
+
+	return kmsapi.NewKeyManagementServiceClient(cnn)
+}
+
+type testService struct {
+	decrypt func(ctx context.Context, uid string, req *DecryptRequest) ([]byte, error)
+	encrypt func(ctx context.Context, uid string, data []byte) (*EncryptResponse, error)
+	status  func(ctx context.Context) (*StatusResponse, error)
+}
+
+var _ Service = (*testService)(nil)
+
+func (s *testService) Decrypt(ctx context.Context, uid string, req *DecryptRequest) ([]byte, error) {
+	return s.decrypt(ctx, uid, req)
+}
+
+func (s *testService) Encrypt(ctx context.Context, uid string, data []byte) (*EncryptResponse, error) {
+	return s.encrypt(ctx, uid, data)
+}
+
+func (s *testService) Status(ctx context.Context) (*StatusResponse, error) {
+	return s.status(ctx)
+}
+
+func makeID(rand func([]byte) (int, error)) (string, error) {
+	b := make([]byte, 10)
+	if _, err := rand(b); err != nil {
+		return "", err
+	}
+
+	return base64.StdEncoding.EncodeToString(b), nil
+}
+
+func newBase64Service(keyID string) *testService {
+	decrypt := func(_ context.Context, _ string, req *DecryptRequest) ([]byte, error) {
+		if req.KeyID != keyID {
+			return nil, fmt.Errorf("keyID mismatch. want: %q, have: %q", keyID, req.KeyID)
+		}
+
+		return base64.StdEncoding.DecodeString(string(req.Ciphertext))
+	}
+
+	encrypt := func(_ context.Context, _ string, data []byte) (*EncryptResponse, error) {
+		return &EncryptResponse{
+			Ciphertext: []byte(base64.StdEncoding.EncodeToString(data)),
+			KeyID:      keyID,
+		}, nil
+	}
+
+	status := func(_ context.Context) (*StatusResponse, error) {
+		return &StatusResponse{
+			Version: version,
+			Healthz: "ok",
+			KeyID:   keyID,
+		}, nil
+	}
+
+	return &testService{
+		decrypt: decrypt,
+		encrypt: encrypt,
+		status:  status,
+	}
+}

staging/src/k8s.io/kms/service/interface.go

@@ -0,0 +1,54 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package service
+
+import "context"
+
+/*
+Copied from: k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2
+*/
+
+// Service allows encrypting and decrypting data using an external Key Management Service.
+type Service interface {
+	// Decrypt a given bytearray to obtain the original data as bytes.
+	Decrypt(ctx context.Context, uid string, req *DecryptRequest) ([]byte, error)
+	// Encrypt bytes to a ciphertext.
+	Encrypt(ctx context.Context, uid string, data []byte) (*EncryptResponse, error)
+	// Status returns the status of the KMS.
+	Status(ctx context.Context) (*StatusResponse, error)
+}
+
+// EncryptResponse is the response from the Envelope service when encrypting data.
+type EncryptResponse struct {
+	Ciphertext  []byte
+	KeyID       string
+	Annotations map[string][]byte
+}
+
+// DecryptRequest is the request to the Envelope service when decrypting data.
+type DecryptRequest struct {
+	Ciphertext  []byte
+	KeyID       string
+	Annotations map[string][]byte
+}
+
+// StatusResponse is the response from the Envelope service when getting the status of the service.
+type StatusResponse struct {
+	Version string
+	Healthz string
+	KeyID   string
+}