Refactor frontend interceptors

chasm/interceptors.go

@@ -5,6 +5,7 @@ import (
 
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
+	interceptornexus "go.temporal.io/server/common/rpc/interceptor/nexus"
 	"google.golang.org/grpc"
 )
 
@@ -60,6 +61,15 @@ func (i *ChasmVisibilityInterceptor) Intercept(
 	return handler(ctx, req)
 }
 
+func (i *ChasmVisibilityInterceptor) InterceptNexus(
+	ctx context.Context,
+	in interceptornexus.InterceptorInput,
+	next interceptornexus.HandlerFunc,
+) (any, error) {
+	ctx = NewVisibilityManagerContext(ctx, i.visibilityMgr)
+	return next(ctx, in)
+}
+
 func ChasmVisibilityInterceptorProvider(visibilityMgr VisibilityManager) *ChasmVisibilityInterceptor {
 	return &ChasmVisibilityInterceptor{
 		visibilityMgr: visibilityMgr,

common/authorization/interceptor.go

@@ -3,8 +3,8 @@ package authorization
 import (
 	"cmp"
 	"context"
-	"crypto/x509"
 	"crypto/x509/pkix"
+	"errors"
 	"time"
 
 	commonpb "go.temporal.io/api/common/v1"
@@ -18,9 +18,11 @@ import (
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
+	commonnexus "go.temporal.io/server/common/nexus"
+	"go.temporal.io/server/common/rpc/interceptor/nexus"
+	"go.temporal.io/server/common/rpc/tlsinfo"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
-	"google.golang.org/grpc/peer"
 )
 
 type (
@@ -54,32 +56,6 @@ var (
 	AuthHeader   contextKeyAuthHeader
 )
 
-// TLSInfoFromContext extracts TLS information from the context's peer value.
-func TLSInfoFromContext(ctx context.Context) *credentials.TLSInfo {
-	p, ok := peer.FromContext(ctx)
-	if !ok {
-		return nil
-	}
-	if tlsInfo, ok := p.AuthInfo.(credentials.TLSInfo); ok {
-		return &tlsInfo
-	}
-	return nil
-}
-
-// PeerCert extracts an x509 certificate from given tlsInfo.
-func PeerCert(tlsInfo *credentials.TLSInfo) *x509.Certificate {
-	if tlsInfo == nil || len(tlsInfo.State.VerifiedChains) == 0 || len(tlsInfo.State.VerifiedChains[0]) == 0 {
-		return nil
-	}
-	// The assumption here is that we only expect a single verified chain of certs (first[0]).
-	// It's unclear how we should handle a situation when more than one chain is presented,
-	// which subject to use. It's okay for us to limit ourselves to one chain.
-	// We can always extend this logic later.
-	// We take the first element in the chain ([0]) because that's the client cert
-	// (at the beginning of the chain), not intermediary CAs or the root CA (at the end of the chain).
-	return tlsInfo.State.VerifiedChains[0][0]
-}
-
 type Interceptor struct {
 	claimMapper                  ClaimMapper
 	authorizer                   Authorizer
@@ -132,7 +108,7 @@ func (a *Interceptor) Intercept(
 	info *grpc.UnaryServerInfo,
 	handler grpc.UnaryHandler,
 ) (any, error) {
-	tlsConnection := TLSInfoFromContext(ctx)
+	tlsConnection := tlsinfo.FromContext(ctx)
 
 	authInfo := a.GetAuthInfo(tlsConnection, headers.NewGRPCHeaderGetter(ctx), func() string {
 		if a.audienceGetter != nil {
@@ -184,6 +160,58 @@ func (a *Interceptor) Intercept(
 	return handler(ctx, req)
 }
 
+func (a *Interceptor) InterceptNexus(
+	ctx context.Context,
+	in nexus.InterceptorInput,
+	next nexus.HandlerFunc,
+) (any, error) {
+	a.logger.Debug("authorizing request")
+	ctx = headers.StripPrincipal(ctx)
+	if a.authorizer == nil {
+		return next(ctx, in)
+	}
+	namespaceName := in.NamespaceName()
+	apiName := in.APIName()
+	endpointName := in.EndpointName()
+	claims, _ := ctx.Value(MappedClaims).(*Claims) //nolint:revive // unchecked-type-assertion: empty claims will 403
+	ct := &CallTarget{
+		APIName:           apiName,
+		NexusEndpointName: endpointName,
+		Namespace:         namespaceName,
+		Request:           in.Request(),
+	}
+	principal, err := a.Authorize(ctx, claims, ct)
+	if err != nil {
+		if permissionDeniedError, ok := errors.AsType[*serviceerror.PermissionDenied](err); ok {
+			a.logger.Debug("Request unauthorized")
+			return nil, &nexus.InterceptorError{
+				Err:                       commonnexus.AdaptAuthorizeError(permissionDeniedError),
+				Outcome:                   "unauthorized",
+				SkipServiceErrorReporting: true,
+			}
+		}
+		logTags := []tag.Tag{
+			tag.Operation(api.MethodName(apiName)),
+			tag.WorkflowNamespace(namespaceName),
+			tag.Endpoint(endpointName),
+			tag.Error(err),
+		}
+		if operationName := in.OperationName(); operationName != "" {
+			logTags = append(logTags, tag.NexusOperation(operationName))
+		}
+		a.logger.Error("Authorization internal error with processing nexus request", logTags...)
+		return nil, &nexus.InterceptorError{
+			Err:                       err,
+			Outcome:                   "internal_auth_error",
+			SkipServiceErrorReporting: true,
+		}
+	}
+	if a.enablePrincipalPropagation != nil && a.enablePrincipalPropagation(namespaceName) && principal != nil {
+		ctx = headers.SetPrincipal(ctx, principal)
+	}
+	return next(ctx, in)
+}
+
 // InterceptStream is a gRPC stream server interceptor that enforces authorization.
 func (a *Interceptor) InterceptStream(
 	srv any,
@@ -194,7 +222,7 @@ func (a *Interceptor) InterceptStream(
 	ctx := ss.Context()
 	bypassAuth := a.disableStreamingAuthorizer()
 	if !bypassAuth {
-		tlsConnection := TLSInfoFromContext(ctx)
+		tlsConnection := tlsinfo.FromContext(ctx)
 		headerGetter := headers.NewGRPCHeaderGetter(ctx)
 
 		authInfo := a.GetAuthInfo(tlsConnection, headerGetter, func() string {
@@ -260,7 +288,7 @@ func (a *Interceptor) GetAuthInfo(tlsConnection *credentials.TLSInfo, header hea
 		authHeader = header.Get(a.authHeaderName)
 		authExtraHeader = header.Get(a.authExtraHeaderName)
 	}
-	clientCert := PeerCert(tlsConnection)
+	clientCert := tlsinfo.PeerCert(tlsConnection)
 	if clientCert != nil {
 		tlsSubject = &clientCert.Subject
 	}

common/authorization/interceptor_test.go

@@ -8,20 +8,24 @@ import (
 	"errors"
 	"slices"
 	"testing"
+	"time"
 
+	"github.com/nexus-rpc/sdk-go/nexus"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 	commandpb "go.temporal.io/api/command/v1"
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/api/workflowservice/v1"
+	"go.temporal.io/server/api/matchingservice/v1"
 	"go.temporal.io/server/common/api"
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/headers"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
+	interceptornexus "go.temporal.io/server/common/rpc/interceptor/nexus"
 	"go.uber.org/mock/gomock"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
@@ -68,6 +72,78 @@ func TestAuthorizerInterceptorSuite(t *testing.T) {
 	suite.Run(t, s)
 }
 
+func (s *authorizerInterceptorSuite) TestInterceptNexus() {
+	apiName, endpoint := "NexusAPI", "endpoint"
+	authorizationRequest := &matchingservice.DispatchNexusTaskRequest{}
+	input := interceptornexus.NewStartOpInput(
+		"s",
+		"o",
+		testNamespace,
+		time.Now(),
+		nexus.StartOperationOptions{},
+		nil,
+		interceptornexus.ForwardingInfo{},
+		interceptornexus.RequestMetadata{
+			APIName:      apiName,
+			EndpointName: endpoint,
+			Request:      authorizationRequest,
+		},
+	)
+	expectedTarget := &CallTarget{
+		APIName:           apiName,
+		NexusEndpointName: endpoint,
+		Namespace:         testNamespace,
+		Request:           authorizationRequest,
+	}
+	for _, tc := range []struct {
+		name                string
+		ctx                 context.Context
+		authorizationResult *Result
+		nextCalled          bool
+		expectedError       error
+	}{
+		{
+			name:                "authorized",
+			ctx:                 context.Background(),
+			authorizationResult: &Result{Decision: DecisionAllow},
+			nextCalled:          true,
+		},
+		{
+			name:                "unauthorized",
+			ctx:                 context.Background(),
+			authorizationResult: &Result{Decision: DecisionDeny},
+			expectedError: &interceptornexus.InterceptorError{
+				Err:                       nexus.NewHandlerErrorf(nexus.HandlerErrorTypeUnauthorized, "permission denied"),
+				Outcome:                   "unauthorized",
+				SkipServiceErrorReporting: true,
+			},
+		},
+	} {
+		s.Run(tc.name, func() {
+			if tc.authorizationResult != nil {
+				s.mockAuthorizer.EXPECT().Authorize(gomock.Any(), nil, expectedTarget).
+					Return(*tc.authorizationResult, nil)
+				if tc.authorizationResult.Decision == DecisionDeny {
+					s.mockMetricsHandler.EXPECT().
+						Counter(metrics.ServiceErrUnauthorizedCounter.Name()).
+						Return(metrics.NoopCounterMetricFunc)
+				}
+			}
+
+			nextCalled := false
+			_, err := s.interceptor.InterceptNexus(
+				tc.ctx,
+				input,
+				func(context.Context, interceptornexus.InterceptorInput) (any, error) {
+					nextCalled = true
+					return nil, nil
+				})
+			s.Equal(tc.expectedError, err)
+			s.Equal(tc.nextCalled, nextCalled)
+		})
+	}
+}
+
 func (s *authorizerInterceptorSuite) SetupTest() {
 	s.Assertions = require.New(s.T())
 	s.controller = gomock.NewController(s.T())

common/rpc/grpcfaults/interceptor.go

@@ -3,6 +3,7 @@ package grpcfaults
 import (
 	"context"
 
+	"go.temporal.io/server/common/rpc/interceptor/nexus"
 	"google.golang.org/grpc"
 )
 
@@ -43,3 +44,33 @@ func UnaryServerInterceptor(generator Generator) grpc.UnaryServerInterceptor {
 		return resp, err
 	}
 }
+
+func NewFaultsInterceptor(generator Generator) *FaultsInterceptor {
+	return &FaultsInterceptor{
+		h: UnaryServerInterceptor(generator),
+	}
+}
+
+type FaultsInterceptor struct {
+	h grpc.UnaryServerInterceptor
+}
+
+func (g *FaultsInterceptor) Intercept(
+	ctx context.Context,
+	req any,
+	info *grpc.UnaryServerInfo,
+	handler grpc.UnaryHandler,
+) (any, error) {
+	if g.h == nil {
+		return handler(ctx, req)
+	}
+	return g.h(ctx, req, info, handler)
+}
+
+func (g *FaultsInterceptor) InterceptNexus(
+	ctx context.Context,
+	in nexus.InterceptorInput,
+	next nexus.HandlerFunc,
+) (any, error) {
+	return next(ctx, in)
+}

common/rpc/interceptor/caller_info.go

@@ -6,6 +6,7 @@ import (
 	"go.temporal.io/server/common/api"
 	"go.temporal.io/server/common/headers"
 	"go.temporal.io/server/common/namespace"
+	"go.temporal.io/server/common/rpc/interceptor/nexus"
 	"google.golang.org/grpc"
 )
 
@@ -40,6 +41,20 @@ func (i *CallerInfoInterceptor) Intercept(
 	return handler(ctx, req)
 }
 
+// InterceptNexus adds caller information for a Nexus request.
+func (i *CallerInfoInterceptor) InterceptNexus(
+	ctx context.Context,
+	in nexus.InterceptorInput,
+	next nexus.HandlerFunc,
+) (any, error) {
+	ctx = PopulateCallerInfo(
+		ctx,
+		in.NamespaceName,
+		in.MethodName,
+	)
+	return next(headers.Propagate(ctx), in)
+}
+
 // PopulateCallerInfo gets current caller info value from the context and updates any that are missing.
 // Namespace name and method are passed as functions to avoid expensive lookups if those values are already set.
 func PopulateCallerInfo(