update the api server/client to handle both authenticated and unauthe…

…nticated client

Signed-off-by: Rewant Soni <[email protected]>

services/provider/client/client.go

@@ -44,6 +44,27 @@ func NewProviderClient(ctx context.Context, serverAddr string, timeout time.Dura
 		timeout:    timeout}, nil
 }
 
+// NewAuthenticatedOCSClient creates a client to talk to the external OCS storage provider server with authentication
+func NewAuthenticatedOCSClient(ctx context.Context, serverAddr string, timeout time.Duration, clientTLSConfig *tls.Config) (*OCSProviderClient, error) {
+	apiCtx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	opts := []grpc.DialOption{
+		grpc.WithTransportCredentials(credentials.NewTLS(clientTLSConfig)),
+		grpc.WithBlock(),
+	}
+
+	conn, err := grpc.DialContext(apiCtx, serverAddr, opts...)
+	if err != nil {
+		return nil, fmt.Errorf("failed to dial: %v", err)
+	}
+
+	return &OCSProviderClient{
+		Client:     pb.NewOCSProviderClient(conn),
+		clientConn: conn,
+		timeout:    timeout}, nil
+}
+
 // Close closes the gRPC connection of the external OCS storage provider client
 func (cc *OCSProviderClient) Close() {
 	if cc.clientConn != nil {

services/provider/server/server.go

@@ -5,13 +5,15 @@ import (
 	"crypto"
 	"crypto/rsa"
 	"crypto/sha256"
+	"crypto/tls"
 	"crypto/x509"
 	"encoding/base64"
 	"encoding/hex"
 	"encoding/json"
 	"encoding/pem"
 	"fmt"
 	"net"
+	"os"
 	"strconv"
 	"strings"
 	"time"
@@ -28,6 +30,7 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/reflection"
 	"google.golang.org/grpc/status"
 	corev1 "k8s.io/api/core/v1"
@@ -203,13 +206,36 @@ func (s *OCSProviderServer) Start(port int, opts []grpc.ServerOption) {
 
 	certFile := ProviderCertsMountPoint + "/tls.crt"
 	keyFile := ProviderCertsMountPoint + "/tls.key"
-	creds, sslErr := credentials.NewServerTLSFromFile(certFile, keyFile)
-	if sslErr != nil {
-		klog.Fatalf("Failed loading certificates: %v", sslErr)
+	caCertFile := ProviderCertsMountPoint + "/ca.crt"
+
+	certificate, err := tls.LoadX509KeyPair(certFile, keyFile)
+	if err != nil {
+		klog.Fatalf("Failed loading certificates: %v", err)
+		return
+	}
+
+	data, err := os.ReadFile(caCertFile)
+	if err != nil {
+		klog.Fatalf("Failed to load CA file: %v", err)
 		return
 	}
 
-	opts = append(opts, grpc.Creds(creds))
+	capool := x509.NewCertPool()
+	if !capool.AppendCertsFromPEM(data) {
+		klog.Fatal("Failed to add CA Cert")
+		return
+	}
+
+	tlsConfig := &tls.Config{
+		ClientAuth:   tls.VerifyClientCertIfGiven,
+		Certificates: []tls.Certificate{certificate},
+		ClientCAs:    capool,
+	}
+
+	creds := credentials.NewTLS(tlsConfig)
+
+	opts = append(opts, grpc.Creds(creds), grpc.UnaryInterceptor(authenticationFilterInterceptor))
+
 	grpcServer := grpc.NewServer(opts...)
 	pb.RegisterOCSProviderServer(grpcServer, s)
 	// Register reflection service on gRPC server.
@@ -220,6 +246,19 @@ func (s *OCSProviderServer) Start(port int, opts []grpc.ServerOption) {
 	}
 }
 
+func authenticationFilterInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
+	if info.FullMethod == "provider.OCSProvider/PeerBlockPool" {
+		if p, ok := peer.FromContext(ctx); ok {
+			if mtls, ok := p.AuthInfo.(credentials.TLSInfo); ok {
+				if len(mtls.State.PeerCertificates) <= 0 {
+					return nil, status.Errorf(codes.Unauthenticated, "the call PeerBlockPool should flow from an authenticated client ")
+				}
+			}
+		}
+	}
+	return handler(ctx, req)
+}
+
 func newClient() (client.Client, error) {
 	scheme := runtime.NewScheme()
 	err := ocsv1alpha1.AddToScheme(scheme)