Update.

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

Author: xuezhaojun

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

@@ -62,10 +62,6 @@ func NewSARAuthorizer(kubeClient kubernetes.Interface) *SARAuthorizer {
 	}
 }
 
-func (s *SARAuthorizer) FullMethodName() string {
-	return pbv1.CloudEventService_Subscribe_FullMethodName
-}
-
 func (s *SARAuthorizer) AuthorizeRequest(ctx context.Context, req any) error {
 	pReq, ok := req.(*pbv1.PublishRequest)
 	if !ok {
@@ -95,6 +91,10 @@ func (s *SARAuthorizer) AuthorizeStream(ctx context.Context, ss grpc.ServerStrea
 		return ss, nil
 	}
 
+	if info.FullMethod != pbv1.CloudEventService_Subscribe_FullMethodName {
+		return nil, fmt.Errorf("unsupported service full method %s for SARAuthorizer", info.FullMethod)
+	}
+
 	var req pbv1.SubscriptionRequest
 	if err := ss.RecvMsg(&req); err != nil {
 		return nil, err

pkg/server/grpc/authz/interface.go

@@ -12,5 +12,4 @@ type UnaryAuthorizer interface {
 
 type StreamAuthorizer interface {
 	AuthorizeStream(ctx context.Context, ss grpc.ServerStream, info *grpc.StreamServerInfo) (grpc.ServerStream, error)
-	FullMethodName() string
 }

pkg/server/grpc/server.go

@@ -24,13 +24,12 @@ type GRPCServer struct {
 	registerFuncs     []func(*grpc.Server)
 	authenticators    []authn.Authenticator
 	unaryAuthorizers  []authz.UnaryAuthorizer
-	streamAuthorizers map[string][]authz.StreamAuthorizer
+	streamAuthorizers []authz.StreamAuthorizer
 }
 
 func NewGRPCServer(opt *GRPCServerOptions) *GRPCServer {
 	return &GRPCServer{
-		options:           opt,
-		streamAuthorizers: make(map[string][]authz.StreamAuthorizer),
+		options: opt,
 	}
 }
 
@@ -50,7 +49,7 @@ func (b *GRPCServer) WithUnaryAuthorizer(authorizer authz.UnaryAuthorizer) *GRPC
 }
 
 func (b *GRPCServer) WithStreamAuthorizer(authorizer authz.StreamAuthorizer) *GRPCServer {
-	b.streamAuthorizers[authorizer.FullMethodName()] = append(b.streamAuthorizers[authorizer.FullMethodName()], authorizer)
+	b.streamAuthorizers = append(b.streamAuthorizers, authorizer)
 	return b
 }
 
@@ -238,7 +237,7 @@ func newAuthnStreamInterceptor(authenticators ...authn.Authenticator) grpc.Strea
 }
 
 // newAuthzStreamInterceptor is a stream interceptor that authorizes the stream request.
-func newAuthzStreamInterceptor(authorizers map[string][]authz.StreamAuthorizer) grpc.StreamServerInterceptor {
+func newAuthzStreamInterceptor(authorizers []authz.StreamAuthorizer) grpc.StreamServerInterceptor {
 	return func(
 		srv interface{},
 		ss grpc.ServerStream,
@@ -247,13 +246,11 @@ func newAuthzStreamInterceptor(authorizers map[string][]authz.StreamAuthorizer)
 	) error {
 		var errs []error
 
-		if authorizers, ok := authorizers[info.FullMethod]; ok {
-			for _, authorizer := range authorizers {
-				if authorizedStream, err := authorizer.AuthorizeStream(ss.Context(), ss, info); err == nil {
-					return handler(srv, authorizedStream)
-				} else {
-					errs = append(errs, err)
-				}
+		for _, authorizer := range authorizers {
+			if authorizedStream, err := authorizer.AuthorizeStream(ss.Context(), ss, info); err == nil {
+				return handler(srv, authorizedStream)
+			} else {
+				errs = append(errs, err)
 			}
 		}