Merge remote-tracking branch 'upstream' into secfixfinal

Author: eshitachandwani

binarylog/grpc_binarylog_v1/binarylog.pb.go

@@ -245,6 +245,10 @@ var (
 	AsyncReporterCleanupDelegate = func(cleanup func()) func() {
 		return cleanup
 	}
+
+	// XDSFilterWrapperOption returns a ServerOption that sets the internal
+	// stream wrapper used for server-side xDS HTTP filters.
+	XDSFilterWrapperOption any // func(func(grpc.ServerStream) (grpc.ServerStream, error)) grpc.ServerOption
 )
 
 // HealthChecker defines the signature of the client-side LB channel health

internal/internal.go

@@ -58,17 +58,6 @@ type RPCConfig struct {
 	Interceptor  any
 }
 
-// ServerInterceptor is an interceptor for incoming RPC's on gRPC server side.
-type ServerInterceptor interface {
-	// AllowRPC checks if an incoming RPC is allowed to proceed based on
-	// information about connection RPC was received on, and HTTP Headers. This
-	// information will be piped into context.
-	AllowRPC(ctx context.Context) error // TODO: Make this a real interceptor for filters such as rate limiting.
-	// Close closes the interceptor. Once called, no new calls to NewStream are
-	// accepted. Ongoing calls to NewStream are allowed to complete.
-	Close()
-}
-
 type csKeyType string
 
 const csKey = csKeyType("grpc.internal.resolver.configSelector")

internal/resolver/config_selector.go

@@ -95,8 +95,8 @@ type ClientInterceptor interface {
 	// Note: RPCInfo.Context is currently unused and will be nil.
 	NewStream(ctx context.Context, ri iresolver.RPCInfo, newStream func(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStream, error), opts ...grpc.CallOption) (grpc.ClientStream, error)
 
-	// Close closes the interceptor. Once called, no new calls to NewStream are
-	// accepted. Ongoing calls to NewStream are allowed to complete.
+	// Close closes the interceptor. No new RPCs will be dispatched to this
+	// interceptor, but ongoing calls to NewStream are allowed to complete.
 	Close()
 }
 
@@ -131,6 +131,25 @@ type ClientFilter interface {
 	Close()
 }
 
+// ServerInterceptor is an interceptor for incoming RPC's on gRPC server side.
+type ServerInterceptor interface {
+	// InterceptRPC intercepts an incoming RPC on the server side.
+	//
+	// On success, implementations must return either the original ServerStream
+	// or a wrapped ServerStream, with a nil error.
+	//
+	// Returning a non-nil error will terminate the RPC with that error.
+	// Implementations are expected to return an error created using the status
+	// package; otherwise, the RPC will fail with an UNKNOWN status code.
+	//
+	// Implementations should never return (nil, nil).
+	InterceptRPC(ss grpc.ServerStream) (grpc.ServerStream, error)
+
+	// Close closes the interceptor. No new RPCs will be dispatched to this
+	// interceptor, but ongoing calls to InterceptRPC are allowed to complete.
+	Close()
+}
+
 // ServerFilterBuilder is an optional interface that a Builder can implement to
 // indicate its capability to build server-side filters.
 type ServerFilterBuilder interface {
@@ -149,7 +168,7 @@ type ServerFilter interface {
 	//
 	// It is valid for this method to return a nil Interceptor and a nil error.
 	// In this case, the RPC will not be intercepted by this filter.
-	BuildServerInterceptor(config, override FilterConfig) (iresolver.ServerInterceptor, error)
+	BuildServerInterceptor(config, override FilterConfig) (ServerInterceptor, error)
 
 	// Close is called when the filter is no longer needed.
 	Close()

internal/xds/httpfilter/httpfilter.go

@@ -20,12 +20,11 @@
 package rbac
 
 import (
-	"context"
 	"errors"
 	"fmt"
 	"strings"
 
-	"google.golang.org/grpc/internal/resolver"
+	"google.golang.org/grpc"
 	"google.golang.org/grpc/internal/xds/httpfilter"
 	"google.golang.org/grpc/internal/xds/rbac"
 	"google.golang.org/protobuf/proto"
@@ -236,7 +235,7 @@ type serverFilter struct{}
 
 func (serverFilter) Close() {}
 
-func (serverFilter) BuildServerInterceptor(cfg httpfilter.FilterConfig, override httpfilter.FilterConfig) (resolver.ServerInterceptor, error) {
+func (serverFilter) BuildServerInterceptor(cfg httpfilter.FilterConfig, override httpfilter.FilterConfig) (httpfilter.ServerInterceptor, error) {
 	if cfg == nil {
 		return nil, fmt.Errorf("rbac: nil config provided")
 	}
@@ -270,8 +269,11 @@ type interceptor struct {
 	chainEngine *rbac.ChainEngine
 }
 
-func (i *interceptor) AllowRPC(ctx context.Context) error {
-	return i.chainEngine.IsAuthorized(ctx)
+func (i *interceptor) InterceptRPC(ss grpc.ServerStream) (grpc.ServerStream, error) {
+	if err := i.chainEngine.IsAuthorized(ss.Context()); err != nil {
+		return nil, err
+	}
+	return ss, nil
 }
 
 func (i *interceptor) Close() {}

internal/xds/httpfilter/rbac/rbac.go

@@ -22,7 +22,6 @@ package router
 import (
 	"fmt"
 
-	iresolver "google.golang.org/grpc/internal/resolver"
 	"google.golang.org/grpc/internal/xds/httpfilter"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/anypb"
@@ -104,7 +103,7 @@ func (filter) BuildClientInterceptor(cfg, override httpfilter.FilterConfig) (htt
 	return nil, nil
 }
 
-func (filter) BuildServerInterceptor(cfg, override httpfilter.FilterConfig) (iresolver.ServerInterceptor, error) {
+func (filter) BuildServerInterceptor(cfg, override httpfilter.FilterConfig) (httpfilter.ServerInterceptor, error) {
 	if _, ok := cfg.(config); !ok {
 		return nil, fmt.Errorf("router: incorrect config type provided (%T): %v", cfg, cfg)
 	}

internal/xds/httpfilter/router/router.go

@@ -18,16 +18,15 @@
 package server
 
 import (
-	"context"
 	"errors"
 	"fmt"
 	"net/netip"
 	"slices"
 	"strings"
 	"sync/atomic"
 
+	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/internal/resolver"
 	"google.golang.org/grpc/internal/xds/httpfilter"
 	"google.golang.org/grpc/internal/xds/xdsclient/xdsresource"
 	"google.golang.org/grpc/status"
@@ -209,7 +208,7 @@ type virtualHostWithInterceptors struct {
 type routeWithInterceptors struct {
 	matcher     *xdsresource.CompositeMatcher
 	actionType  xdsresource.RouteActionType
-	interceptor resolver.ServerInterceptor
+	interceptor httpfilter.ServerInterceptor
 }
 
 type lookupParams struct {
@@ -500,9 +499,9 @@ func (rc *usableRouteConfiguration) statusErrWithNodeID(c codes.Code, msg string
 	return status.Error(c, fmt.Sprintf("[xDS node id: %v]: %s", rc.nodeID, fmt.Sprintf(msg, args...)))
 }
 
-func (fc *filterChain) newInterceptor(routeOverride, virtualHostOverride map[string]httpfilter.FilterConfig, provider serverFilterProvider) (_ resolver.ServerInterceptor, _ []httpfilter.ServerFilter, err error) {
+func (fc *filterChain) newInterceptor(routeOverride, virtualHostOverride map[string]httpfilter.FilterConfig, provider serverFilterProvider) (_ httpfilter.ServerInterceptor, _ []httpfilter.ServerFilter, err error) {
 	serverFilters := []httpfilter.ServerFilter{}
-	interceptors := make([]resolver.ServerInterceptor, 0, len(fc.httpFilters))
+	interceptors := make([]httpfilter.ServerInterceptor, 0, len(fc.httpFilters))
 	defer func() {
 		if err != nil {
 			for _, sf := range serverFilters {
@@ -549,16 +548,18 @@ func (fc *filterChain) newInterceptor(routeOverride, virtualHostOverride map[str
 }
 
 type interceptorList struct {
-	interceptors []resolver.ServerInterceptor
+	interceptors []httpfilter.ServerInterceptor
 }
 
-func (il *interceptorList) AllowRPC(ctx context.Context) error {
+func (il *interceptorList) InterceptRPC(ss grpc.ServerStream) (grpc.ServerStream, error) {
+	var err error
 	for _, i := range il.interceptors {
-		if err := i.AllowRPC(ctx); err != nil {
-			return err
+		ss, err = i.InterceptRPC(ss)
+		if err != nil {
+			return nil, err
 		}
 	}
-	return nil
+	return ss, nil
 }
 
 func (il *interceptorList) Close() {

internal/xds/server/filter_chain_manager.go

@@ -18,7 +18,6 @@
 package server
 
 import (
-	"context"
 	"errors"
 	"fmt"
 	"net/netip"
@@ -28,8 +27,8 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
+	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/tls/certprovider"
-	iresolver "google.golang.org/grpc/internal/resolver"
 	"google.golang.org/grpc/internal/testutils"
 	"google.golang.org/grpc/internal/testutils/xds/e2e"
 	"google.golang.org/grpc/internal/xds/bootstrap"
@@ -569,7 +568,7 @@ func (fb *filterBuilder) Close() {}
 
 var _ httpfilter.ServerFilterBuilder = &filterBuilder{}
 
-func (fb *filterBuilder) BuildServerInterceptor(config httpfilter.FilterConfig, override httpfilter.FilterConfig) (iresolver.ServerInterceptor, error) {
+func (fb *filterBuilder) BuildServerInterceptor(config httpfilter.FilterConfig, override httpfilter.FilterConfig) (httpfilter.ServerInterceptor, error) {
 	var level string
 	level = config.(filterCfg).level
 
@@ -583,8 +582,8 @@ type serverInterceptor struct {
 	level string
 }
 
-func (si *serverInterceptor) AllowRPC(context.Context) error {
-	return errors.New(si.level)
+func (si *serverInterceptor) InterceptRPC(grpc.ServerStream) (grpc.ServerStream, error) {
+	return nil, errors.New(si.level)
 }
 
 func (si *serverInterceptor) Close() {}
@@ -699,8 +698,6 @@ func (s) TestHTTPFilterInstantiation(t *testing.T) {
 			wantErrs: []string{topLevel, vhLevel, rLevel},
 		},
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			fc := filterChain{
@@ -727,7 +724,8 @@ func (s) TestHTTPFilterInstantiation(t *testing.T) {
 			var errs []string
 			for _, vh := range urc.vhs {
 				for _, r := range vh.routes {
-					errs = append(errs, r.interceptor.AllowRPC(ctx).Error())
+					_, err := r.interceptor.InterceptRPC(nil)
+					errs = append(errs, err.Error())
 				}
 			}
 			if !cmp.Equal(errs, test.wantErrs) {

internal/xds/server/filter_chain_manager_test.go

@@ -19,27 +19,25 @@
 package server
 
 import (
-	"context"
 	"errors"
-	"fmt"
 	"strings"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/internal/transport"
 	"google.golang.org/grpc/internal/xds/xdsclient/xdsresource"
 	"google.golang.org/grpc/metadata"
-	"google.golang.org/grpc/status"
 )
 
 // RouteAndProcess routes the incoming RPC to a configured route in the route
 // table and also processes the RPC by running the incoming RPC through any HTTP
 // Filters configured.
-func RouteAndProcess(ctx context.Context) error {
+func RouteAndProcess(ss grpc.ServerStream) (grpc.ServerStream, error) {
+	ctx := ss.Context()
 	conn := transport.GetConnection(ctx)
 	cw, ok := conn.(*connWrapper)
 	if !ok {
-		return errors.New("missing virtual hosts in incoming context")
+		return nil, errors.New("missing virtual hosts in incoming context")
 	}
 
 	rc := cw.urc.Load()
@@ -50,28 +48,28 @@ func RouteAndProcess(ctx context.Context) error {
 		if logger.V(2) {
 			logger.Infof("RPC on connection with xDS Configuration error: %v", rc.err)
 		}
-		return status.Error(codes.Unavailable, fmt.Sprintf("error from xDS configuration for matched route configuration: %v", rc.err))
+		return nil, rc.statusErrWithNodeID(codes.Unavailable, "error from xDS configuration for matched route configuration: %v", rc.err)
 	}
 
 	mn, ok := grpc.Method(ctx)
 	if !ok {
-		return errors.New("missing method name in incoming context")
+		return nil, rc.statusErrWithNodeID(codes.Internal, "missing method name in incoming context")
 	}
 	md, ok := metadata.FromIncomingContext(ctx)
 	if !ok {
-		return errors.New("missing metadata in incoming context")
+		return nil, rc.statusErrWithNodeID(codes.Internal, "missing metadata in incoming context")
 	}
 	// A41 added logic to the core grpc implementation to guarantee that once the
 	// RPC gets to this point, there will be a single, unambiguous authority
 	// present in the header map. But add a defensive check to ensure authority
 	// header is present.
 	authority := md.Get(":authority")
 	if len(authority) == 0 {
-		return rc.statusErrWithNodeID(codes.Internal, "no :authority header present")
+		return nil, rc.statusErrWithNodeID(codes.Internal, "no :authority header present")
 	}
 	vh := findBestMatchingVirtualHostServer(authority[0], rc.vhs)
 	if vh == nil {
-		return rc.statusErrWithNodeID(codes.Unavailable, "the incoming RPC did not match a configured Virtual Host")
+		return nil, rc.statusErrWithNodeID(codes.Unavailable, "the incoming RPC did not match a configured Virtual Host")
 	}
 
 	var rwi *routeWithInterceptors
@@ -81,19 +79,19 @@ func RouteAndProcess(ctx context.Context) error {
 			// server-side; a route with an inappropriate action causes RPCs
 			// matching that route to fail with UNAVAILABLE." - A36
 			if r.actionType != xdsresource.RouteActionNonForwardingAction {
-				return rc.statusErrWithNodeID(codes.Unavailable, "the incoming RPC matched to a route that was not of action type non forwarding")
+				return nil, rc.statusErrWithNodeID(codes.Unavailable, "the incoming RPC matched to a route that was not of action type non forwarding")
 			}
 			rwi = &r
 			break
 		}
 	}
 	if rwi == nil {
-		return rc.statusErrWithNodeID(codes.Unavailable, "the incoming RPC did not match a configured Route")
+		return nil, rc.statusErrWithNodeID(codes.Unavailable, "the incoming RPC did not match a configured Route")
 	}
-	if err := rwi.interceptor.AllowRPC(ctx); err != nil {
-		return rc.statusErrWithNodeID(codes.PermissionDenied, "Incoming RPC is not allowed: %v", err)
+	if rwi.interceptor != nil {
+		return rwi.interceptor.InterceptRPC(ss)
 	}
-	return nil
+	return ss, nil
 }
 
 // findBestMatchingVirtualHostServer returns the virtual host whose domains field best

internal/xds/server/routing.go

@@ -125,6 +125,15 @@ func (s *testServerTransportStream) SetHeader(metadata.MD) error  { return nil }
 func (s *testServerTransportStream) SendHeader(metadata.MD) error { return nil }
 func (s *testServerTransportStream) SetTrailer(metadata.MD) error { return nil }
 
+type testServerStream struct {
+	grpc.ServerStream
+	ctx context.Context
+}
+
+func (s *testServerStream) Context() context.Context {
+	return s.ctx
+}
+
 func (s) TestRouteAndProcess_MissingAuthority(t *testing.T) {
 	var ptr atomic.Pointer[usableRouteConfiguration]
 	ptr.Store(&usableRouteConfiguration{})
@@ -135,7 +144,8 @@ func (s) TestRouteAndProcess_MissingAuthority(t *testing.T) {
 	ctx = grpc.NewContextWithServerTransportStream(ctx, &testServerTransportStream{method: "/test.Service/Method"})
 	ctx = metadata.NewIncomingContext(ctx, metadata.MD{})
 
-	err := RouteAndProcess(ctx)
+	ss := &testServerStream{ctx: ctx}
+	_, err := RouteAndProcess(ss)
 	if status.Code(err) != codes.Internal {
 		t.Fatalf("RouteAndProcess() returned error code %v, want %v", status.Code(err), codes.Internal)
 	}