Add option to disable auth on streaming endpoint

Author: yux0

common/authorization/interceptor.go

@@ -89,6 +89,7 @@ type Interceptor struct {
 	authHeaderName               string
 	authExtraHeaderName          string
 	enableCrossNamespaceCommands dynamicconfig.BoolPropertyFn
+	disableStreamingAuthorizer   dynamicconfig.BoolPropertyFn
 }
 
 // NewInterceptor creates an authorization interceptor.
@@ -102,6 +103,7 @@ func NewInterceptor(
 	authHeaderName string,
 	authExtraHeaderName string,
 	enableCrossNamespaceCommands dynamicconfig.BoolPropertyFn,
+	disableStreamingAuthorizer dynamicconfig.BoolPropertyFn,
 ) *Interceptor {
 	return &Interceptor{
 		claimMapper:                  claimMapper,
@@ -113,6 +115,7 @@ func NewInterceptor(
 		authExtraHeaderName:          cmp.Or(authExtraHeaderName, defaultAuthExtraHeaderName),
 		audienceGetter:               audienceGetter,
 		enableCrossNamespaceCommands: enableCrossNamespaceCommands,
+		disableStreamingAuthorizer:   disableStreamingAuthorizer,
 	}
 }
 
@@ -174,34 +177,37 @@ func (a *Interceptor) InterceptStream(
 	handler grpc.StreamHandler,
 ) error {
 	ctx := ss.Context()
-	tlsConnection := TLSInfoFromContext(ctx)
-
-	authInfo := a.GetAuthInfo(tlsConnection, headers.NewGRPCHeaderGetter(ctx), func() string {
-		// JWTAudienceMapper only supports UnaryServerInfo; no request is available at stream init.
-		return ""
-	})
-
-	var claims *Claims
-	if authInfo != nil {
-		var err error
-		claims, err = a.GetClaims(authInfo)
-		if err != nil {
-			a.logger.Error("Authorization error", tag.Error(err))
-			return errUnauthorized
+	bypassAuth := a.disableStreamingAuthorizer()
+	if !bypassAuth {
+		tlsConnection := TLSInfoFromContext(ctx)
+
+		authInfo := a.GetAuthInfo(tlsConnection, headers.NewGRPCHeaderGetter(ctx), func() string {
+			// JWTAudienceMapper only supports UnaryServerInfo; no request is available at stream init.
+			return ""
+		})
+
+		var claims *Claims
+		if authInfo != nil {
+			var err error
+			claims, err = a.GetClaims(authInfo)
+			if err != nil {
+				a.logger.Error("Authorization error", tag.Error(err))
+				return errUnauthorized
+			}
+			ctx = a.EnhanceContext(ctx, authInfo, claims)
 		}
-		ctx = a.EnhanceContext(ctx, authInfo, claims)
-	}
 
-	if a.authorizer != nil {
-		// Namespace is not available in the stream handshake (no initial request body).
-		ct := &CallTarget{
-			Namespace: "",
-			APIName:   info.FullMethod,
-			Request:   nil,
-		}
-		if err := a.Authorize(ctx, claims, ct); err != nil {
-			a.logger.Error("Authorization error", tag.Error(err))
-			return err
+		if a.authorizer != nil {
+			// Namespace is not available in the stream handshake (no initial request body).
+			ct := &CallTarget{
+				Namespace: "",
+				APIName:   info.FullMethod,
+				Request:   nil,
+			}
+			if err := a.Authorize(ctx, claims, ct); err != nil {
+				a.logger.Error("Authorization error", tag.Error(err))
+				return err
+			}
 		}
 	}
 

common/authorization/interceptor_test.go

@@ -82,6 +82,7 @@ func (s *authorizerInterceptorSuite) SetupTest() {
 		"",
 		"",
 		dynamicconfig.GetBoolPropertyFn(false), // enableCrossNamespaceCommands
+		dynamicconfig.GetBoolPropertyFn(false), // disableStreamingAuthorizer
 	)
 	s.handler = func(ctx context.Context, req interface{}) (interface{}, error) { return true, nil }
 }
@@ -165,6 +166,7 @@ func (s *authorizerInterceptorSuite) TestNoopClaimMapperWithoutTLS() {
 		"",
 		"",
 		dynamicconfig.GetBoolPropertyFn(false), // enableCrossNamespaceCommands
+		dynamicconfig.GetBoolPropertyFn(false), // disableStreamingAuthorizer
 	)
 	_, err := interceptor.Intercept(ctx, describeNamespaceRequest, describeNamespaceInfo, s.handler)
 	s.NoError(err)
@@ -181,6 +183,7 @@ func (s *authorizerInterceptorSuite) TestAlternateHeaders() {
 		"custom-header",
 		"custom-extra-header",
 		dynamicconfig.GetBoolPropertyFn(false), // enableCrossNamespaceCommands
+		dynamicconfig.GetBoolPropertyFn(false), // disableStreamingAuthorizer
 	)
 
 	cases := []struct {
@@ -288,7 +291,8 @@ func (s *authorizerInterceptorSuite) newCrossNamespaceInterceptor(namespaces ...
 		nil,
 		"",
 		"",
-		dynamicconfig.GetBoolPropertyFn(true), // enableCrossNamespaceCommands
+		dynamicconfig.GetBoolPropertyFn(true),  // enableCrossNamespaceCommands
+		dynamicconfig.GetBoolPropertyFn(false), // disableStreamingAuthorizer
 	)
 }
 
@@ -605,6 +609,7 @@ func (s *authorizerInterceptorSuite) TestInterceptStream_AuthDisabled() {
 		"",
 		"",
 		dynamicconfig.GetBoolPropertyFn(false),
+		dynamicconfig.GetBoolPropertyFn(false),
 	)
 
 	handlerCalled := false
@@ -630,6 +635,7 @@ func (s *authorizerInterceptorSuite) TestInterceptStream_InvalidToken() {
 		"",
 		"",
 		dynamicconfig.GetBoolPropertyFn(false),
+		dynamicconfig.GetBoolPropertyFn(false),
 	)
 
 	// Provide an incoming context with an auth token so GetAuthInfo returns non-nil.

common/dynamicconfig/constants.go

@@ -148,6 +148,11 @@ for signal / start / signal with start API if namespace is not active`,
 		true,
 		`EnableCrossNamespaceCommands is the key to enable commands for external namespaces`,
 	)
+	DisableStreamingAuthorizer = NewGlobalBoolSetting(
+		"system.disableStreamingAuthorizer",
+		false,
+		`DisableStreamingAuthorizer is the key to disable the auth on streaming endpoint`,
+	)
 	ClusterMetadataRefreshInterval = NewGlobalDurationSetting(
 		"system.clusterMetadataRefreshInterval",
 		time.Minute,

service/frontend/fx.go

@@ -171,6 +171,7 @@ func AuthorizationInterceptorProvider(
 		cfg.Global.Authorization.AuthHeaderName,
 		cfg.Global.Authorization.AuthExtraHeaderName,
 		dynamicconfig.EnableCrossNamespaceCommands.Get(dc),
+		dynamicconfig.DisableStreamingAuthorizer.Get(dc),
 	)
 }
 

service/frontend/nexus_handler_test.go

@@ -128,6 +128,7 @@ func newOperationContext(options contextOptions) *operationContext {
 		"",
 		"",
 		dynamicconfig.GetBoolPropertyFn(false), // enableCrossNamespaceCommands
+		dynamicconfig.GetBoolPropertyFn(false), // disableStreamingAuthorizer
 	)
 	oc.namespaceConcurrencyLimitInterceptor = interceptor.NewConcurrentRequestLimitInterceptor(
 		nil,