grpc
/
grpc-go
mirror of https://github.com/grpc/grpc-go.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

xds: Add env var protection for RBAC HTTP Filter (#4765) 

* xds: Add env var protection for RBAC HTTP Filter
This commit is contained in:
Zach Reyes 2021-09-17 01:01:07 -04:00 committed by GitHub
parent 567da6b863
commit e469f0d5f5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 322 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
internal/xds/env/env.go vendored
View File

@ -43,6 +43,7 @@ const (
clientSideSecuritySupportEnv = "GRPC_XDS_EXPERIMENTAL_SECURITY_SUPPORT"
aggregateAndDNSSupportEnv = "GRPC_XDS_EXPERIMENTAL_ENABLE_AGGREGATE_AND_LOGICAL_DNS_CLUSTER"
retrySupportEnv = "GRPC_XDS_EXPERIMENTAL_ENABLE_RETRY"
rbacSupportEnv = "GRPC_XDS_EXPERIMENTAL_ENABLE_RBAC"
c2pResolverSupportEnv = "GRPC_EXPERIMENTAL_GOOGLE_C2P_RESOLVER"
c2pResolverTestOnlyTrafficDirectorURIEnv = "GRPC_TEST_ONLY_GOOGLE_C2P_RESOLVER_TRAFFIC_DIRECTOR_URI"
@ -82,6 +83,9 @@ var (
// RetrySupport indicates whether xDS retry is enabled.
RetrySupport = !strings.EqualFold(os.Getenv(retrySupportEnv), "false")
// RBACSupport indicates whether xDS configured RBAC HTTP Filter is enabled.
RBACSupport = strings.EqualFold(os.Getenv(rbacSupportEnv), "true")
// C2PResolverSupport indicates whether support for C2P resolver is enabled.
// This can be enabled by setting the environment variable
// "GRPC_EXPERIMENTAL_GOOGLE_C2P_RESOLVER" to "true".

5
xds/internal/httpfilter/httpfilter.go
View File

@ -94,6 +94,11 @@ func Register(b Filter) {
}
}
// UnregisterForTesting unregisters the HTTP Filter for testing purposes.
func UnregisterForTesting(typeURL string) {
delete(m, typeURL)
}
// Get returns the HTTPFilter registered with typeURL.
//
// If no filter is register with typeURL, nil will be returned.

19
xds/internal/httpfilter/rbac/rbac.go
View File

@ -28,6 +28,7 @@ import (
"github.com/golang/protobuf/proto"
"github.com/golang/protobuf/ptypes"
"google.golang.org/grpc/internal/resolver"
"google.golang.org/grpc/internal/xds/env"
"google.golang.org/grpc/internal/xds/rbac"
"google.golang.org/grpc/xds/internal/httpfilter"
"google.golang.org/protobuf/types/known/anypb"
@ -37,9 +38,27 @@ import (
)
func init() {
if env.RBACSupport {
httpfilter.Register(builder{})
}
}
// RegisterForTesting registers the RBAC HTTP Filter for testing purposes, regardless
// of the RBAC environment variable. This is needed because there is no way to set the RBAC
// environment variable to true in a test before init() in this package is run.
func RegisterForTesting() {
httpfilter.Register(builder{})
}
// UnregisterForTesting unregisters the RBAC HTTP Filter for testing purposes. This is needed because
// there is no way to unregister the HTTP Filter after registering it solely for testing purposes using
// rbac.RegisterForTesting()
func UnregisterForTesting() {
for _, typeURL := range builder.TypeURLs(builder{}) {
httpfilter.UnregisterForTesting(typeURL)
}
}
type builder struct {
}

10
xds/internal/server/listener_wrapper.go
View File

@ -35,6 +35,7 @@ import (
internalbackoff "google.golang.org/grpc/internal/backoff"
internalgrpclog "google.golang.org/grpc/internal/grpclog"
"google.golang.org/grpc/internal/grpcsync"
"google.golang.org/grpc/internal/xds/env"
"google.golang.org/grpc/xds/internal/xdsclient"
"google.golang.org/grpc/xds/internal/xdsclient/bootstrap"
)
@ -272,6 +273,9 @@ func (l *listenerWrapper) Accept() (net.Conn, error) {
conn.Close()
continue
}
if !env.RBACSupport {
return &connWrapper{Conn: conn, filterChain: fc, parent: l}, nil
}
var rc xdsclient.RouteConfigUpdate
if fc.InlineRouteConfig != nil {
rc = *fc.InlineRouteConfig
@ -410,8 +414,10 @@ func (l *listenerWrapper) handleLDSUpdate(update ldsUpdateWithError) {
// Server's state to ServingModeNotServing. That prevents new connections
// from being accepted, whereas here we simply want the clients to reconnect
// to get the updated configuration.
if l.drainCallback != nil {
l.drainCallback(l.Listener.Addr())
if env.RBACSupport {
if l.drainCallback != nil {
l.drainCallback(l.Listener.Addr())
}
}
l.rdsHandler.updateRouteNamesToWatch(ilc.FilterChains.RouteConfigNames)
// If there are no dynamic RDS Configurations still needed to be received

6
xds/internal/server/listener_wrapper_test.go
View File

@ -34,6 +34,7 @@ import (
wrapperspb "github.com/golang/protobuf/ptypes/wrappers"
"google.golang.org/grpc/internal/grpctest"
"google.golang.org/grpc/internal/testutils"
"google.golang.org/grpc/internal/xds/env"
_ "google.golang.org/grpc/xds/internal/httpfilter/router"
"google.golang.org/grpc/xds/internal/testutils/e2e"
"google.golang.org/grpc/xds/internal/testutils/fakeclient"
@ -325,6 +326,11 @@ func (s) TestNewListenerWrapper(t *testing.T) {
// the update from the rds handler should it move the server to
// ServingModeServing.
func (s) TestNewListenerWrapperWithRouteUpdate(t *testing.T) {
oldRBAC := env.RBACSupport
env.RBACSupport = true
defer func() {
env.RBACSupport = oldRBAC
}()
_, readyCh, xdsC, _, cleanup := newListenerWrapper(t)
defer cleanup()

226
xds/internal/test/xds_server_integration_test.go
View File

@ -34,9 +34,10 @@ import (
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/internal/testutils"
"google.golang.org/grpc/internal/xds/env"
"google.golang.org/grpc/status"
"google.golang.org/grpc/xds"
_ "google.golang.org/grpc/xds/internal/httpfilter/rbac"
"google.golang.org/grpc/xds/internal/httpfilter/rbac"
"google.golang.org/grpc/xds/internal/testutils/e2e"
v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
@ -374,6 +375,11 @@ func (s) TestServerSideXDS_SecurityConfigChange(t *testing.T) {
// (NonForwardingAction), and the RPC's matching those routes should proceed as
// normal.
func (s) TestServerSideXDS_RouteConfiguration(t *testing.T) {
oldRBAC := env.RBACSupport
env.RBACSupport = true
defer func() {
env.RBACSupport = oldRBAC
}()
managementServer, nodeID, bootstrapContents, resolver, cleanup1 := setupManagementServer(t)
defer cleanup1()
@ -711,6 +717,13 @@ func serverListenerWithRBACHTTPFilters(host string, port uint32, rbacCfg *rpb.RB
// as normal and certain RPC's are denied by the RBAC HTTP Filter which gets
// called by hooked xds interceptors.
func (s) TestRBACHTTPFilter(t *testing.T) {
oldRBAC := env.RBACSupport
env.RBACSupport = true
defer func() {
env.RBACSupport = oldRBAC
}()
rbac.RegisterForTesting()
defer rbac.UnregisterForTesting()
tests := []struct {
name string
rbacCfg *rpb.RBAC
@ -823,7 +836,218 @@ func (s) TestRBACHTTPFilter(t *testing.T) {
if _, err := client.UnaryCall(ctx, &testpb.SimpleRequest{}); status.Code(err) != test.wantStatusUnaryCall {
t.Fatalf("UnaryCall() returned err with status: %v, wantStatusUnaryCall: %v", err, test.wantStatusUnaryCall)
}
// Toggle the RBAC Env variable off, this should disable RBAC and allow any RPC"s through (will not go through
// routing or processed by HTTP Filters and thus will never get denied by RBAC).
env.RBACSupport = false
if _, err := client.EmptyCall(ctx, &testpb.Empty{}); status.Code(err) != codes.OK {
t.Fatalf("EmptyCall() returned err with status: %v, once RBAC is disabled all RPC's should proceed as normal", status.Code(err))
}
if _, err := client.UnaryCall(ctx, &testpb.SimpleRequest{}); status.Code(err) != codes.OK {
t.Fatalf("UnaryCall() returned err with status: %v, once RBAC is disabled all RPC's should proceed as normal", status.Code(err))
}
// Toggle RBAC back on for next iterations.
env.RBACSupport = true
}()
})
}
}
// serverListenerWithBadRouteConfiguration returns an xds Listener resource with
// a Route Configuration that will never successfully match in order to test
// RBAC Environment variable being toggled on and off.
func serverListenerWithBadRouteConfiguration(host string, port uint32) *v3listenerpb.Listener {
return &v3listenerpb.Listener{
Name: fmt.Sprintf(e2e.ServerListenerResourceNameTemplate, net.JoinHostPort(host, strconv.Itoa(int(port)))),
Address: &v3corepb.Address{
Address: &v3corepb.Address_SocketAddress{
SocketAddress: &v3corepb.SocketAddress{
Address: host,
PortSpecifier: &v3corepb.SocketAddress_PortValue{
PortValue: port,
},
},
},
},
FilterChains: []*v3listenerpb.FilterChain{
{
Name: "v4-wildcard",
FilterChainMatch: &v3listenerpb.FilterChainMatch{
PrefixRanges: []*v3corepb.CidrRange{
{
AddressPrefix: "0.0.0.0",
PrefixLen: &wrapperspb.UInt32Value{
Value: uint32(0),
},
},
},
SourceType: v3listenerpb.FilterChainMatch_SAME_IP_OR_LOOPBACK,
SourcePrefixRanges: []*v3corepb.CidrRange{
{
AddressPrefix: "0.0.0.0",
PrefixLen: &wrapperspb.UInt32Value{
Value: uint32(0),
},
},
},
},
Filters: []*v3listenerpb.Filter{
{
Name: "filter-1",
ConfigType: &v3listenerpb.Filter_TypedConfig{
TypedConfig: testutils.MarshalAny(&v3httppb.HttpConnectionManager{
RouteSpecifier: &v3httppb.HttpConnectionManager_RouteConfig{
RouteConfig: &v3routepb.RouteConfiguration{
Name: "routeName",
VirtualHosts: []*v3routepb.VirtualHost{{
// Incoming RPC's will try and match to Virtual Hosts based on their :authority header.
// Thus, incoming RPC's will never match to a Virtual Host (server side requires matching
// to a VH/Route of type Non Forwarding Action to proceed normally), and all incoming RPC's
// with this route configuration will be denied.
Domains: []string{"will-never-match"},
Routes: []*v3routepb.Route{{
Match: &v3routepb.RouteMatch{
PathSpecifier: &v3routepb.RouteMatch_Prefix{Prefix: "/"},
},
Action: &v3routepb.Route_NonForwardingAction{},
}}}}},
},
HttpFilters: []*v3httppb.HttpFilter{e2e.RouterHTTPFilter},
}),
},
},
},
},
{
Name: "v6-wildcard",
FilterChainMatch: &v3listenerpb.FilterChainMatch{
PrefixRanges: []*v3corepb.CidrRange{
{
AddressPrefix: "::",
PrefixLen: &wrapperspb.UInt32Value{
Value: uint32(0),
},
},
},
SourceType: v3listenerpb.FilterChainMatch_SAME_IP_OR_LOOPBACK,
SourcePrefixRanges: []*v3corepb.CidrRange{
{
AddressPrefix: "::",
PrefixLen: &wrapperspb.UInt32Value{
Value: uint32(0),
},
},
},
},
Filters: []*v3listenerpb.Filter{
{
Name: "filter-1",
ConfigType: &v3listenerpb.Filter_TypedConfig{
TypedConfig: testutils.MarshalAny(&v3httppb.HttpConnectionManager{
RouteSpecifier: &v3httppb.HttpConnectionManager_RouteConfig{
RouteConfig: &v3routepb.RouteConfiguration{
Name: "routeName",
VirtualHosts: []*v3routepb.VirtualHost{{
// Incoming RPC's will try and match to Virtual Hosts based on their :authority header.
// Thus, incoming RPC's will never match to a Virtual Host (server side requires matching
// to a VH/Route of type Non Forwarding Action to proceed normally), and all incoming RPC's
// with this route configuration will be denied.
Domains: []string{"will-never-match"},
Routes: []*v3routepb.Route{{
Match: &v3routepb.RouteMatch{
PathSpecifier: &v3routepb.RouteMatch_Prefix{Prefix: "/"},
},
Action: &v3routepb.Route_NonForwardingAction{},
}}}}},
},
HttpFilters: []*v3httppb.HttpFilter{e2e.RouterHTTPFilter},
}),
},
},
},
},
},
}
}
// TestRBACToggledOffThenToggledOnWithBadRouteConfiguration tests a scenario
// where the server gets a listener configuration with a route table that is
// garbage, with incoming RPC's never matching to a VH/Route of type Non
// Forwarding Action, thus never proceeding as normal. In the default scenario
// (RBAC Env Var turned off, thus all logic related to Route Configuration
// protected), the RPC's should simply proceed as normal due to ignoring the
// route configuration. Once toggling the route configuration on, the RPC's
// should all fail after updating the Server.
func (s) TestRBACToggledOffThenToggledOnWithBadRouteConfiguration(t *testing.T) {
managementServer, nodeID, bootstrapContents, resolver, cleanup1 := setupManagementServer(t)
defer cleanup1()
lis, cleanup2 := setupGRPCServer(t, bootstrapContents)
defer cleanup2()
host, port, err := hostPortFromListener(lis)
if err != nil {
t.Fatalf("failed to retrieve host and port of server: %v", err)
}
const serviceName = "my-service-fallback"
// The inbound listener needs a route table that will never match on a VH,
// and thus shouldn't allow incoming RPC's to proceed.
resources := e2e.DefaultClientResources(e2e.ResourceParams{
DialTarget: serviceName,
NodeID: nodeID,
Host: host,
Port: port,
SecLevel: e2e.SecurityLevelNone,
})
// This bad route configuration shouldn't affect incoming RPC's from
// proceeding as normal, as the configuration shouldn't be parsed due to the
// RBAC Environment variable not being set to true.
inboundLis := serverListenerWithBadRouteConfiguration(host, port)
resources.Listeners = append(resources.Listeners, inboundLis)
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
// Setup the management server with client and server-side resources.
if err := managementServer.Update(ctx, resources); err != nil {
t.Fatal(err)
}
cc, err := grpc.DialContext(ctx, fmt.Sprintf("xds:///%s", serviceName), grpc.WithInsecure(), grpc.WithResolvers(resolver))
if err != nil {
t.Fatalf("failed to dial local test server: %v", err)
}
defer cc.Close()
client := testpb.NewTestServiceClient(cc)
// The default setting of RBAC being disabled should allow any RPC's to
// proceed as normal.
if _, err := client.EmptyCall(ctx, &testpb.Empty{}); status.Code(err) != codes.OK {
t.Fatalf("EmptyCall() returned err with status: %v, if RBAC is disabled all RPC's should proceed as normal", status.Code(err))
}
if _, err := client.UnaryCall(ctx, &testpb.SimpleRequest{}); status.Code(err) != codes.OK {
t.Fatalf("UnaryCall() returned err with status: %v, if RBAC is disabled all RPC's should proceed as normal", status.Code(err))
}
// After toggling RBAC support on, all the RPC's should get denied with
// status code Unavailable due to not matching to a route of type Non
// Forwarding Action (Route Table not configured properly).
oldRBAC := env.RBACSupport
env.RBACSupport = true
defer func() {
env.RBACSupport = oldRBAC
}()
// Update the server with the same configuration, this is blocking on server
// side so no raciness here.
if err := managementServer.Update(ctx, resources); err != nil {
t.Fatal(err)
}
if _, err := client.EmptyCall(ctx, &testpb.Empty{}); status.Code(err) != codes.Unavailable {
t.Fatalf("EmptyCall() returned err with status: %v, if RBAC is disabled all RPC's should proceed as normal", status.Code(err))
}
if _, err := client.UnaryCall(ctx, &testpb.SimpleRequest{}); status.Code(err) != codes.Unavailable {
t.Fatalf("UnaryCall() returned err with status: %v, if RBAC is disabled all RPC's should proceed as normal", status.Code(err))
}
}

4
xds/internal/xdsclient/filter_chain.go
View File

@ -29,6 +29,7 @@ import (
"github.com/golang/protobuf/proto"
"github.com/golang/protobuf/ptypes"
"google.golang.org/grpc/internal/resolver"
"google.golang.org/grpc/internal/xds/env"
"google.golang.org/grpc/xds/internal/httpfilter"
"google.golang.org/grpc/xds/internal/version"
)
@ -598,6 +599,9 @@ func processNetworkFilters(filters []*v3listenerpb.Filter) (*FilterChain, error)
// TODO: Implement terminal filter logic, as per A36.
filterChain.HTTPFilters = filters
seenHCM = true
if !env.RBACSupport {
continue
}
switch hcm.RouteSpecifier.(type) {
case *v3httppb.HttpConnectionManager_Rds:
if hcm.GetRds().GetConfigSource().GetAds() == nil {

36
xds/internal/xdsclient/filter_chain_test.go
View File

@ -40,6 +40,7 @@ import (
iresolver "google.golang.org/grpc/internal/resolver"
"google.golang.org/grpc/internal/testutils"
"google.golang.org/grpc/internal/xds/env"
"google.golang.org/grpc/xds/internal/httpfilter"
"google.golang.org/grpc/xds/internal/httpfilter/router"
"google.golang.org/grpc/xds/internal/testutils/e2e"
@ -519,6 +520,11 @@ func TestNewFilterChainImpl_Failure_BadSecurityConfig(t *testing.T) {
// TestNewFilterChainImpl_Success_RouteUpdate tests the construction of the
// filter chain with valid HTTP Filters present.
func TestNewFilterChainImpl_Success_RouteUpdate(t *testing.T) {
oldRBAC := env.RBACSupport
env.RBACSupport = true
defer func() {
env.RBACSupport = oldRBAC
}()
tests := []struct {
name string
lis *v3listenerpb.Listener
@ -754,6 +760,11 @@ func TestNewFilterChainImpl_Success_RouteUpdate(t *testing.T) {
// TestNewFilterChainImpl_Failure_BadRouteUpdate verifies cases where the Route
// Update in the filter chain are invalid.
func TestNewFilterChainImpl_Failure_BadRouteUpdate(t *testing.T) {
oldRBAC := env.RBACSupport
env.RBACSupport = true
defer func() {
env.RBACSupport = oldRBAC
}()
tests := []struct {
name string
lis *v3listenerpb.Listener
@ -927,6 +938,11 @@ func TestNewFilterChainImpl_Failure_BadHTTPFilters(t *testing.T) {
// TestNewFilterChainImpl_Success_HTTPFilters tests the construction of the
// filter chain with valid HTTP Filters present.
func TestNewFilterChainImpl_Success_HTTPFilters(t *testing.T) {
oldRBAC := env.RBACSupport
env.RBACSupport = true
defer func() {
env.RBACSupport = oldRBAC
}()
tests := []struct {
name string
lis *v3listenerpb.Listener
@ -1245,6 +1261,11 @@ func TestNewFilterChainImpl_Success_HTTPFilters(t *testing.T) {
// TestNewFilterChainImpl_Success_SecurityConfig verifies cases where the
// security configuration in the filter chain contains valid data.
func TestNewFilterChainImpl_Success_SecurityConfig(t *testing.T) {
oldRBAC := env.RBACSupport
env.RBACSupport = true
defer func() {
env.RBACSupport = oldRBAC
}()
tests := []struct {
desc string
lis *v3listenerpb.Listener
@ -1472,6 +1493,11 @@ func TestNewFilterChainImpl_Success_SecurityConfig(t *testing.T) {
// success at config validation time and the filter chains which contains
// unsupported match fields will be skipped at lookup time.
func TestNewFilterChainImpl_Success_UnsupportedMatchFields(t *testing.T) {
oldRBAC := env.RBACSupport
env.RBACSupport = true
defer func() {
env.RBACSupport = oldRBAC
}()
unspecifiedEntry := &destPrefixEntry{
srcTypeArr: [3]*sourcePrefixes{
{
@ -1637,6 +1663,11 @@ func TestNewFilterChainImpl_Success_UnsupportedMatchFields(t *testing.T) {
// TestNewFilterChainImpl_Success_AllCombinations verifies different
// combinations of the supported match criteria.
func TestNewFilterChainImpl_Success_AllCombinations(t *testing.T) {
oldRBAC := env.RBACSupport
env.RBACSupport = true
defer func() {
env.RBACSupport = oldRBAC
}()
tests := []struct {
desc string
lis *v3listenerpb.Listener
@ -2283,6 +2314,11 @@ func TestLookup_Failures(t *testing.T) {
}
func TestLookup_Successes(t *testing.T) {
oldRBAC := env.RBACSupport
env.RBACSupport = true
defer func() {
env.RBACSupport = oldRBAC
}()
lisWithDefaultChain := &v3listenerpb.Listener{
FilterChains: []*v3listenerpb.FilterChain{
{

6
xds/internal/xdsclient/lds_test.go
View File

@ -33,6 +33,7 @@ import (
"google.golang.org/protobuf/types/known/durationpb"
"google.golang.org/grpc/internal/testutils"
"google.golang.org/grpc/internal/xds/env"
"google.golang.org/grpc/xds/internal/httpfilter"
_ "google.golang.org/grpc/xds/internal/httpfilter/router"
"google.golang.org/grpc/xds/internal/testutils/e2e"
@ -601,6 +602,11 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) {
}
func (s) TestUnmarshalListener_ServerSide(t *testing.T) {
oldRBAC := env.RBACSupport
env.RBACSupport = true
defer func() {
env.RBACSupport = oldRBAC
}()
const (
v3LDSTarget = "grpc/server?xds.resource.listening_address=0.0.0.0:9999"
testVersion = "test-version-lds-server"

13
xds/server.go
View File

@ -37,6 +37,7 @@ import (
"google.golang.org/grpc/internal/grpcsync"
iresolver "google.golang.org/grpc/internal/resolver"
"google.golang.org/grpc/internal/transport"
"google.golang.org/grpc/internal/xds/env"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
"google.golang.org/grpc/xds/internal/server"
@ -381,8 +382,10 @@ func routeAndProcess(ctx context.Context) error {
// xdsUnaryInterceptor is the unary interceptor added to the gRPC server to
// perform any xDS specific functionality on unary RPCs.
func xdsUnaryInterceptor(ctx context.Context, req interface{}, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
if err := routeAndProcess(ctx); err != nil {
return nil, err
if env.RBACSupport {
if err := routeAndProcess(ctx); err != nil {
return nil, err
}
}
return handler(ctx, req)
}
@ -390,8 +393,10 @@ func xdsUnaryInterceptor(ctx context.Context, req interface{}, _ *grpc.UnaryServ
// xdsStreamInterceptor is the stream interceptor added to the gRPC server to
// perform any xDS specific functionality on streaming RPCs.
func xdsStreamInterceptor(srv interface{}, ss grpc.ServerStream, _ *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
if err := routeAndProcess(ss.Context()); err != nil {
return err
if env.RBACSupport {
if err := routeAndProcess(ss.Context()); err != nil {
return err
}
}
return handler(srv, ss)
}