Refactor grpc pkg to support general GRPC service.

Signed-off-by: xuezhaojun <zxue@redhat.com>

Author: xuezhaojun

pkg/cloudevents/server/grpc/authn/interface.go

@@ -3,6 +3,11 @@ package sar
 import (
 	"context"
 	"fmt"
+	"sync"
+
+	"google.golang.org/grpc"
+	pbv1 "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc/protobuf/v1"
+	"open-cluster-management.io/sdk-go/pkg/server/grpc/authz"
 
 	authv1 "k8s.io/api/authorization/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -16,41 +21,123 @@ import (
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/clients/lease"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work/payload"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
-	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/authn"
-	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/authz"
+	"open-cluster-management.io/sdk-go/pkg/server/grpc/authn"
 )
 
 type SARAuthorizer struct {
 	kubeClient kubernetes.Interface
 }
 
-func NewSARAuthorizer(kubeClient kubernetes.Interface) authz.Authorizer {
+// validate SARAuthorizer implement StreamAuthorizer and UnaryAuthorizer
+var _ authz.StreamAuthorizer = (*SARAuthorizer)(nil)
+var _ authz.UnaryAuthorizer = (*SARAuthorizer)(nil)
+
+// wrappedAuthorizedStream caches the subscription request that is already read.
+type wrappedAuthorizedStream struct {
+	sync.Mutex
+
+	grpc.ServerStream
+	authorizedReq *pbv1.SubscriptionRequest
+}
+
+// RecvMsg set the msg from the cache.
+func (c *wrappedAuthorizedStream) RecvMsg(m any) error {
+	c.Lock()
+	defer c.Unlock()
+
+	msg, ok := m.(*pbv1.SubscriptionRequest)
+	if !ok {
+		return fmt.Errorf("unsupported request type %T", m)
+	}
+
+	msg.ClusterName = c.authorizedReq.ClusterName
+	msg.Source = c.authorizedReq.Source
+	msg.DataType = c.authorizedReq.DataType
+	return nil
+}
+
+func NewSARAuthorizer(kubeClient kubernetes.Interface) *SARAuthorizer {
 	return &SARAuthorizer{
 		kubeClient: kubeClient,
 	}
 }
 
-func (s *SARAuthorizer) Authorize(ctx context.Context, cluster string, eventsType types.CloudEventsType) error {
+func (s *SARAuthorizer) AuthorizeRequest(ctx context.Context, req any) (authz.Decision, error) {
+	pReq, ok := req.(*pbv1.PublishRequest)
+	if !ok {
+		return authz.DecisionDeny, fmt.Errorf("unsupported request type %T", req)
+	}
+
+	eventsType, err := types.ParseCloudEventsType(pReq.Event.Type)
+	if err != nil {
+		return authz.DecisionDeny, err
+	}
+
+	// the event of grpc publish request is the original cloudevent data, we need a `ce-` prefix
+	// to get the event attribute
+	clusterAttr, ok := pReq.Event.Attributes[fmt.Sprintf("ce-%s", types.ExtensionClusterName)]
+	if !ok {
+		return authz.DecisionDeny, fmt.Errorf("missing ce-clustername in event attributes, %v", pReq.Event.Attributes)
+	}
+
+	decision, err := s.authorize(ctx, clusterAttr.GetCeString(), *eventsType)
+	return decision, err
+}
+
+func (s *SARAuthorizer) AuthorizeStream(ctx context.Context, ss grpc.ServerStream, info *grpc.StreamServerInfo) (authz.Decision, grpc.ServerStream, error) {
+	if info.IsClientStream {
+		return authz.DecisionAllow, ss, nil
+	}
+
+	if info.FullMethod != pbv1.CloudEventService_Subscribe_FullMethodName {
+		return authz.DecisionDeny, nil, fmt.Errorf("unsupported service full method %s for SARAuthorizer", info.FullMethod)
+	}
+
+	var req pbv1.SubscriptionRequest
+	if err := ss.RecvMsg(&req); err != nil {
+		return authz.DecisionDeny, nil, err
+	}
+
+	eventDataType, err := types.ParseCloudEventsDataType(req.DataType)
+	if err != nil {
+		return authz.DecisionDeny, nil, err
+	}
+
+	eventsType := types.CloudEventsType{
+		CloudEventsDataType: *eventDataType,
+		SubResource:         types.SubResourceSpec,
+		Action:              types.WatchRequestAction,
+	}
+
+	decision, err := s.authorize(ss.Context(), req.ClusterName, eventsType)
+	if err != nil {
+		return decision, nil, err
+	}
+
+	return decision, &wrappedAuthorizedStream{ServerStream: ss, authorizedReq: &req}, nil
+}
+
+func (s *SARAuthorizer) authorize(ctx context.Context, cluster string, eventsType types.CloudEventsType) (authz.Decision, error) {
 	user, groups, err := userInfo(ctx)
 	if err != nil {
-		return err
+		return authz.DecisionDeny, err
 	}
 
 	sar, err := toSubjectAccessReview(cluster, user, groups, eventsType)
 	if err != nil {
-		return err
+		return authz.DecisionDeny, err
 	}
 
 	created, err := s.kubeClient.AuthorizationV1().SubjectAccessReviews().Create(
 		ctx, sar, metav1.CreateOptions{})
 	if err != nil {
-		return err
+		return authz.DecisionDeny, err
 	}
 	if !created.Status.Allowed {
-		return fmt.Errorf("the event %s is not allowed, (cluster=%s, sar=%v, reason=%v)",
+		return authz.DecisionDeny, fmt.Errorf("the event %s is not allowed, (cluster=%s, sar=%v, reason=%v)",
 			eventsType, cluster, sar.Spec, created.Status)
 	}
-	return nil
+	return authz.DecisionAllow, nil
 }
 
 func userInfo(ctx context.Context) (user string, groups []string, err error) {

pkg/cloudevents/server/grpc/authz/interface.go

@@ -17,7 +17,8 @@ import (
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/clients/lease"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work/payload"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
-	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/authn"
+	"open-cluster-management.io/sdk-go/pkg/server/grpc/authn"
+	"open-cluster-management.io/sdk-go/pkg/server/grpc/authz"
 )
 
 func TestSARAuthorize(t *testing.T) {
@@ -197,13 +198,19 @@ func TestSARAuthorize(t *testing.T) {
 
 			auth := NewSARAuthorizer(client)
 
-			err := auth.Authorize(tc.userCtx(), tc.cluster, tc.eventsType)
+			decision, err := auth.authorize(tc.userCtx(), tc.cluster, tc.eventsType)
 			if tc.expectErr && err == nil {
 				t.Errorf("expected error, got nil")
 			}
 			if !tc.expectErr && err != nil {
 				t.Errorf("unexpected error: %v", err)
 			}
+			if !tc.expectErr && decision != authz.DecisionAllow {
+				t.Errorf("expected DecisionAllow, got %v", decision)
+			}
+			if tc.expectErr && decision != authz.DecisionDeny {
+				t.Errorf("expected DecisionDeny, got %v", decision)
+			}
 		})
 	}
 }

pkg/cloudevents/server/grpc/authz/kube/sar.go

@@ -4,18 +4,15 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"net"
 	"sync"
 	"time"
 
 	"k8s.io/apimachinery/pkg/api/errors"
-	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/cloudevents/sdk-go/v2/binding"
 	cloudeventstypes "github.com/cloudevents/sdk-go/v2/types"
 	"github.com/google/uuid"
-	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/emptypb"
@@ -44,20 +41,17 @@ var _ server.AgentEventServer = &GRPCBroker{}
 // It broadcasts resource spec to agents and listens for resource status updates from them.
 type GRPCBroker struct {
 	pbv1.UnimplementedCloudEventServiceServer
-	grpcServer  *grpc.Server
 	services    map[types.CloudEventsDataType]server.Service
 	subscribers map[string]*subscriber // registered subscribers
 	mu          sync.RWMutex
 }
 
 // NewGRPCBroker creates a new gRPC broker with the given gRPC server.
-func NewGRPCBroker(srv *grpc.Server) server.AgentEventServer {
+func NewGRPCBroker() *GRPCBroker {
 	broker := &GRPCBroker{
-		grpcServer:  srv,
 		subscribers: make(map[string]*subscriber),
 		services:    make(map[types.CloudEventsDataType]server.Service),
 	}
-	pbv1.RegisterCloudEventServiceServer(broker.grpcServer, broker)
 	return broker
 }
 
@@ -78,26 +72,6 @@ func (bkr *GRPCBroker) Subscribers() sets.Set[string] {
 	return subscribers
 }
 
-// Start starts the gRPC broker at the given address
-func (bkr *GRPCBroker) Start(ctx context.Context, addr string) {
-	logger := klog.FromContext(ctx)
-	logger.Info("Starting gRPC broker at addr", "addr", addr)
-	lis, err := net.Listen("tcp", addr)
-	if err != nil {
-		utilruntime.Must(fmt.Errorf("failed to listen: %v", err))
-	}
-	go func() {
-		if err := bkr.grpcServer.Serve(lis); err != nil {
-			utilruntime.Must(fmt.Errorf("failed to serve gRPC broker: %v", err))
-		}
-	}()
-
-	// wait until context is canceled
-	<-ctx.Done()
-	klog.Infof("Shutting down gRPC broker")
-	bkr.grpcServer.GracefulStop()
-}
-
 // Publish in stub implementation for agent publish resource status.
 func (bkr *GRPCBroker) Publish(ctx context.Context, pubReq *pbv1.PublishRequest) (*emptypb.Empty, error) {
 	logger := klog.FromContext(ctx)

pkg/cloudevents/server/grpc/authz/kube/sar_test.go

@@ -3,11 +3,13 @@ package grpc
 import (
 	"context"
 	"errors"
+	"net"
 	"testing"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"google.golang.org/grpc"
 	grpccli "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc"
+	pbv1 "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc/protobuf/v1"
 	cetypes "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/server"
 )
@@ -59,17 +61,34 @@ func (s *testService) create(evt *cloudevents.Event) error {
 func TestServer(t *testing.T) {
 	grpcServerOptions := []grpc.ServerOption{}
 	grpcServer := grpc.NewServer(grpcServerOptions...)
-	grpcEventServer := NewGRPCBroker(grpcServer)
+	defer grpcServer.Stop()
+
+	grpcEventServer := NewGRPCBroker()
+	pbv1.RegisterCloudEventServiceServer(grpcServer, grpcEventServer)
 
 	svc := &testService{evts: make(map[string]*cloudevents.Event)}
 	grpcEventServer.RegisterService(dataType, svc)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	go grpcEventServer.Start(ctx, ":8888")
+
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("failed to listen: %v", err)
+	}
+	t.Cleanup(func() {
+		grpcServer.GracefulStop()
+		_ = lis.Close()
+	})
+
+	go func() {
+		if err := grpcServer.Serve(lis); err != nil {
+			t.Errorf("failed to serve: %v", err)
+		}
+	}()
 
 	grpcClientOptions := grpccli.NewGRPCOptions()
-	grpcClientOptions.Dialer = &grpccli.GRPCDialer{URL: "localhost:8888"}
+	grpcClientOptions.Dialer = &grpccli.GRPCDialer{URL: lis.Addr().String()}
 	agentOption := grpccli.NewAgentOptions(grpcClientOptions, "cluster1", "agent1")
 	protocol, err := agentOption.CloudEventsOptions.Protocol(ctx, dataType)
 	if err != nil {
@@ -90,10 +109,10 @@ func TestServer(t *testing.T) {
 		WithResourceID("test2").
 		WithClusterName("cluster1").NewEvent()
 
-	recivedEventCh := make(chan cloudevents.Event)
+	receivedEventCh := make(chan cloudevents.Event)
 	go func() {
 		if err := cloudEventsClient.StartReceiver(ctx, func(event cloudevents.Event) {
-			recivedEventCh <- event
+			receivedEventCh <- event
 		}); err != nil {
 			t.Error(err)
 		}
@@ -110,8 +129,8 @@ func TestServer(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	recievedEvent := <-recivedEventCh
-	if recievedEvent.ID() != evt2.ID() {
+	receivedEvent := <-receivedEventCh
+	if receivedEvent.ID() != evt2.ID() {
 		t.Error("received event is different")
 	}
 }