diff --git a/internal/xds/bootstrap/bootstrap.go b/internal/xds/bootstrap/bootstrap.go index f866da0e8e74..3b398c708531 100644 --- a/internal/xds/bootstrap/bootstrap.go +++ b/internal/xds/bootstrap/bootstrap.go @@ -35,6 +35,7 @@ import ( "google.golang.org/grpc/credentials/tls/certprovider" "google.golang.org/grpc/internal" "google.golang.org/grpc/internal/envconfig" + "google.golang.org/grpc/internal/xds/grpcservice/creds" "google.golang.org/grpc/xds/bootstrap" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/structpb" @@ -132,12 +133,13 @@ type AllowedGRPCService struct { // callCredsConfigs is the list of call-credential configs from the // bootstrap JSON. Kept for Equal and MarshalJSON. callCredsConfigs []CallCredsConfig - // selectedChannelCreds is the first channel-creds entry whose type the - // client supports; it is the one used to build the side channel. - selectedChannelCreds ChannelCreds - // dialOptions are built from the selected channel and call credentials - // and passed to grpc.NewClient when creating the side channel. - dialOptions []grpc.DialOption + // sideChannelCreds is the credentials bundle built from the first + // channel-creds entry whose type the client supports, paired with its + // identity. + sideChannelCreds *creds.ChannelCreds + // sideCallCreds are the call credentials built from the supported + // call-creds configs, paired with their identities, preserving order. + sideCallCreds []*creds.CallCreds // cleanups release resources (credential bundles, file watchers) built // for this service; run when the owning Config is no longer needed. cleanups []func() @@ -148,10 +150,13 @@ func (a *AllowedGRPCService) TargetURI() string { return a.targetURI } -// DialOptions returns the dial options built from this service's selected -// channel and call credentials, for use when creating the side channel. -func (a *AllowedGRPCService) DialOptions() []grpc.DialOption { - return a.dialOptions +// SideChannelCredentials returns the channel and call credentials configured +// for this service, paired with their identities, for use when creating the +// side channel to it. The returned credentials are owned by the bootstrap +// config: their cleanups are nil, and the underlying resources are released +// via Cleanups when the config is no longer needed. +func (a *AllowedGRPCService) SideChannelCredentials() (*creds.ChannelCreds, []*creds.CallCreds) { + return a.sideChannelCreds, a.sideCallCreds } // Cleanups returns cleanups to run when the service is no longer needed. @@ -237,8 +242,10 @@ func (a *AllowedGRPCService) UnmarshalJSON(data []byte) (err error) { } }() - var credsDialOption grpc.DialOption - var selectedChannelCreds ChannelCreds + // The built credentials are paired with their (JSON) identities but the + // pairs carry no cleanups: the resources built here are owned by the + // bootstrap config and released via the cleanups collected below. + var sideChannelCreds *creds.ChannelCreds for _, cc := range jsonS.ChannelCreds { c := bootstrap.GetChannelCredentials(cc.Type) if c == nil { @@ -248,19 +255,18 @@ func (a *AllowedGRPCService) UnmarshalJSON(data []byte) (err error) { if err != nil { return fmt.Errorf("xds: failed to build credentials bundle from bootstrap for allowed grpc service: type %q, err: %v", cc.Type, err) } - selectedChannelCreds = cc - credsDialOption = grpc.WithCredentialsBundle(bundle) + sideChannelCreds = creds.NewChannelCreds(bundle, creds.NewJSONIdentity(cc.Type, cc.Config), nil) cleanups = append(cleanups, cancel) break } - // If no channel-creds type in the list was supported, credsDialOption is + // If no channel-creds type in the list was supported, sideChannelCreds is // still nil after the loop; that is a validation error. - if credsDialOption == nil { + if sideChannelCreds == nil { return fmt.Errorf("xds: no supported channel credentials found for allowed grpc service in config:\n%s", string(data)) } - dialOptions := []grpc.DialOption{credsDialOption} + var sideCallCreds []*creds.CallCreds for _, cfg := range jsonS.CallCredsConfigs { c := bootstrap.GetCallCredentials(cfg.Type) if c == nil { @@ -270,14 +276,14 @@ func (a *AllowedGRPCService) UnmarshalJSON(data []byte) (err error) { if err != nil { return fmt.Errorf("xds: failed to build call credentials from bootstrap for allowed grpc service: type %q, err: %v", cfg.Type, err) } - dialOptions = append(dialOptions, grpc.WithPerRPCCredentials(callCreds)) + sideCallCreds = append(sideCallCreds, creds.NewCallCreds(callCreds, creds.NewJSONIdentity(cfg.Type, cfg.Config), nil)) cleanups = append(cleanups, cancel) } a.channelCreds = jsonS.ChannelCreds a.callCredsConfigs = jsonS.CallCredsConfigs - a.selectedChannelCreds = selectedChannelCreds - a.dialOptions = dialOptions + a.sideChannelCreds = sideChannelCreds + a.sideCallCreds = sideCallCreds a.cleanups = cleanups return nil } @@ -643,6 +649,12 @@ func (c *Config) AllowedGRPCServices() AllowedGRPCServices { return c.allowedGRPCServices } +// AllowedGRPCService returns the allowed gRPC service configured for the +// given target URI, or nil if there is none. +func (c *Config) AllowedGRPCService(targetURI string) *AllowedGRPCService { + return c.allowedGRPCServices[targetURI] +} + // XDSServers returns the top-level list of management servers to connect to, // ordered by priority. func (c *Config) XDSServers() ServerConfigs { diff --git a/internal/xds/bootstrap/bootstrap_test.go b/internal/xds/bootstrap/bootstrap_test.go index e1754ac80135..0fa4bbccf404 100644 --- a/internal/xds/bootstrap/bootstrap_test.go +++ b/internal/xds/bootstrap/bootstrap_test.go @@ -35,6 +35,7 @@ import ( "google.golang.org/grpc/internal/envconfig" "google.golang.org/grpc/internal/grpctest" "google.golang.org/grpc/internal/testutils" + "google.golang.org/grpc/internal/xds/grpcservice/creds" "google.golang.org/grpc/xds/bootstrap" "google.golang.org/protobuf/testing/protocmp" "google.golang.org/protobuf/types/known/structpb" @@ -1742,10 +1743,10 @@ func (s) TestAllowedGRPCServices_UnmarshalJSON(t *testing.T) { name string json string want *AllowedGRPCService - // Fields deliberately excluded from Equal: the selected channel - // creds and the dial options built from the credentials. + // The built credentials are deliberately excluded from Equal; verify + // them via SideChannelCredentials instead. wantSelectedChannelCredsType string - wantDialOptions int + wantSideCallCreds int }{ { name: "insecure_channel_creds", @@ -1755,7 +1756,7 @@ func (s) TestAllowedGRPCServices_UnmarshalJSON(t *testing.T) { channelCreds: []ChannelCreds{{Type: "insecure"}}, }, wantSelectedChannelCredsType: "insecure", - wantDialOptions: 1, + wantSideCallCreds: 0, }, { name: "with_call_creds", @@ -1769,9 +1770,9 @@ func (s) TestAllowedGRPCServices_UnmarshalJSON(t *testing.T) { }}, }, wantSelectedChannelCredsType: "insecure", - // One channel-creds dial option plus one per-RPC call-creds - // option. - wantDialOptions: 2, + // One call credential is built for the supported call-creds + // config. + wantSideCallCreds: 1, }, { name: "unsupported_call_creds_skipped", @@ -1785,8 +1786,8 @@ func (s) TestAllowedGRPCServices_UnmarshalJSON(t *testing.T) { }, wantSelectedChannelCredsType: "insecure", // Unsupported call-creds types are skipped without error, so - // only the channel-creds dial option is built. - wantDialOptions: 1, + // no call credentials are built. + wantSideCallCreds: 0, }, { name: "multiple_supported_call_creds", @@ -1806,9 +1807,9 @@ func (s) TestAllowedGRPCServices_UnmarshalJSON(t *testing.T) { }, }, wantSelectedChannelCredsType: "insecure", - // One channel-creds dial option plus one per-RPC option for - // each supported call credential. - wantDialOptions: 3, + // One call credential is built for each supported call-creds + // config. + wantSideCallCreds: 2, }, { name: "tls_channel_creds", @@ -1818,7 +1819,7 @@ func (s) TestAllowedGRPCServices_UnmarshalJSON(t *testing.T) { channelCreds: []ChannelCreds{{Type: "tls", Config: json.RawMessage("{}")}}, }, wantSelectedChannelCredsType: "tls", - wantDialOptions: 1, + wantSideCallCreds: 0, }, { name: "skips_unsupported_channel_creds", @@ -1828,7 +1829,7 @@ func (s) TestAllowedGRPCServices_UnmarshalJSON(t *testing.T) { channelCreds: []ChannelCreds{{Type: "unsupported_cred_type"}, {Type: "insecure"}}, }, wantSelectedChannelCredsType: "insecure", - wantDialOptions: 1, + wantSideCallCreds: 0, }, } @@ -1845,11 +1846,25 @@ func (s) TestAllowedGRPCServices_UnmarshalJSON(t *testing.T) { if !svc.Equal(test.want) { t.Errorf("parsed service = %+v, want %+v", svc, test.want) } - if got := svc.selectedChannelCreds.Type; got != test.wantSelectedChannelCredsType { - t.Errorf("selectedChannelCreds.Type = %q, want %q", got, test.wantSelectedChannelCredsType) + chanCreds, callCreds := svc.SideChannelCredentials() + if chanCreds == nil || chanCreds.Bundle() == nil { + t.Error("SideChannelCredentials() returned no built channel credentials") } - if got := len(svc.DialOptions()); got != test.wantDialOptions { - t.Errorf("len(DialOptions()) = %d, want %d", got, test.wantDialOptions) + // The identity of the built channel credentials must match the + // first supported channel-creds entry from the bootstrap JSON. + var wantConfig json.RawMessage + for _, cc := range test.want.channelCreds { + if cc.Type == test.wantSelectedChannelCredsType { + wantConfig = cc.Config + break + } + } + wantIdentity := creds.NewChannelCreds(nil, creds.NewJSONIdentity(test.wantSelectedChannelCredsType, wantConfig), nil) + if !chanCreds.Equal(wantIdentity) { + t.Errorf("SideChannelCredentials() channel credentials identity = %+v, want type %q", chanCreds, test.wantSelectedChannelCredsType) + } + if got := len(callCreds); got != test.wantSideCallCreds { + t.Errorf("len(SideChannelCredentials() call creds) = %d, want %d", got, test.wantSideCallCreds) } }) } diff --git a/internal/xds/grpcservice/accesstokencreds/call_creds.go b/internal/xds/grpcservice/accesstokencreds/call_creds.go new file mode 100644 index 000000000000..895977ea79f6 --- /dev/null +++ b/internal/xds/grpcservice/accesstokencreds/call_creds.go @@ -0,0 +1,71 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +// Package accesstokencreds implements static access token CallCredentials for +// xDS-configured side channels, as specified in gRFC A102. +package accesstokencreds + +import ( + "context" + "encoding/json" + "fmt" + + "google.golang.org/grpc/credentials" +) + +// NewCallCredentials returns call credentials that attach a static bearer +// token to outgoing RPCs. The config must be a JSON object of the form +// {"token": }. +// +// The credentials require transport security: the token is only ever sent on +// connections that provide privacy and integrity, and RPCs on weaker +// connections fail. +func NewCallCredentials(configJSON json.RawMessage) (credentials.PerRPCCredentials, error) { + var cfg struct { + Token string `json:"token"` + } + if err := json.Unmarshal(configJSON, &cfg); err != nil { + return nil, fmt.Errorf("failed to unmarshal access token call credentials config: %v", err) + } + if cfg.Token == "" { + return nil, fmt.Errorf("token is required in access token call credentials config") + } + return &callCreds{token: cfg.Token}, nil +} + +// callCreds implements credentials.PerRPCCredentials by attaching a static +// bearer token to each RPC. +type callCreds struct { + token string +} + +// GetRequestMetadata returns the token as an authorization header. It fails +// if the connection does not provide privacy and integrity. +func (c *callCreds) GetRequestMetadata(ctx context.Context, _ ...string) (map[string]string, error) { + ri, _ := credentials.RequestInfoFromContext(ctx) + if err := credentials.CheckSecurityLevel(ri.AuthInfo, credentials.PrivacyAndIntegrity); err != nil { + return nil, fmt.Errorf("unable to transfer access token PerRPCCredentials: %v", err) + } + return map[string]string{"authorization": "Bearer " + c.token}, nil +} + +// RequireTransportSecurity indicates whether the credentials requires +// transport security. +func (c *callCreds) RequireTransportSecurity() bool { + return true +} diff --git a/internal/xds/grpcservice/accesstokencreds/call_creds_test.go b/internal/xds/grpcservice/accesstokencreds/call_creds_test.go new file mode 100644 index 000000000000..3d6f98a322dd --- /dev/null +++ b/internal/xds/grpcservice/accesstokencreds/call_creds_test.go @@ -0,0 +1,121 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package accesstokencreds + +import ( + "context" + "encoding/json" + "testing" + "time" + + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/internal/grpctest" +) + +type s struct { + grpctest.Tester +} + +func Test(t *testing.T) { + grpctest.RunSubTests(t, s{}) +} + +func (s) TestNewCallCredentialsWithInvalidConfig(t *testing.T) { + tests := []struct { + name string + config string + }{ + { + name: "not_an_object", + config: `""`, + }, + { + name: "empty_config", + config: `{}`, + }, + { + name: "empty_token", + config: `{"token": ""}`, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + callCreds, err := NewCallCredentials(json.RawMessage(tt.config)) + if err == nil { + t.Fatalf("NewCallCredentials(%s): got nil, want error", tt.config) + } + if callCreds != nil { + t.Errorf("NewCallCredentials(%s): returned non-nil call credentials", tt.config) + } + }) + } +} + +// Tests that the token is attached as a bearer authorization header on +// connections providing privacy and integrity, and that an error is returned +// on weaker connections. +func (s) TestGetRequestMetadata(t *testing.T) { + const config = `{"token": "test-token"}` + callCreds, err := NewCallCredentials(json.RawMessage(config)) + if err != nil { + t.Fatalf("NewCallCredentials(%s) failed: %v", config, err) + } + + if !callCreds.RequireTransportSecurity() { + t.Error("RequireTransportSecurity() = false, want true") + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + // The token must be attached on a connection with privacy and integrity. + secureCtx := credentials.NewContextWithRequestInfo(ctx, credentials.RequestInfo{ + AuthInfo: &testAuthInfo{secLevel: credentials.PrivacyAndIntegrity}, + }) + md, err := callCreds.GetRequestMetadata(secureCtx) + if err != nil { + t.Fatalf("GetRequestMetadata() on a secure connection failed: %v", err) + } + if got, want := md["authorization"], "Bearer test-token"; got != want { + t.Fatalf("GetRequestMetadata() on a secure connection returned authorization header %q, want %q", got, want) + } + + // An error must be returned on a connection that does not provide + // privacy and integrity. + insecureCtx := credentials.NewContextWithRequestInfo(ctx, credentials.RequestInfo{ + AuthInfo: &testAuthInfo{secLevel: credentials.NoSecurity}, + }) + if md, err := callCreds.GetRequestMetadata(insecureCtx); err == nil { + t.Fatalf("GetRequestMetadata() on an insecure connection returned metadata %v, want error", md) + } +} + +// testAuthInfo implements credentials.AuthInfo for testing. +type testAuthInfo struct { + secLevel credentials.SecurityLevel +} + +func (t *testAuthInfo) AuthType() string { + return "test" +} + +func (t *testAuthInfo) GetCommonAuthInfo() credentials.CommonAuthInfo { + return credentials.CommonAuthInfo{SecurityLevel: t.secLevel} +} diff --git a/internal/xds/grpcservice/creds/creds.go b/internal/xds/grpcservice/creds/creds.go new file mode 100644 index 000000000000..77299ab7a256 --- /dev/null +++ b/internal/xds/grpcservice/creds/creds.go @@ -0,0 +1,171 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +// Package creds defines credentials for xDS-configured side channels: built +// channel and call credentials paired with the identity of the configuration +// they were built from (gRFC A102). +// +// Credentials may be sourced from the bootstrap file (JSON) or from a +// GrpcService proto delivered by a trusted xDS server; the identity captures +// which, and is used to decide whether two configurations may share a +// channel. +package creds + +import ( + "bytes" + "encoding/json" + "sync" + + "google.golang.org/grpc/credentials" + "google.golang.org/protobuf/types/known/anypb" +) + +// Identity identifies the configuration a credential was built from. It has +// two flavors, bootstrap JSON and GrpcService proto, and is used only for +// equality comparisons, never as a map key. +type Identity interface { + // Type returns the credential type this identity describes: the + // bootstrap credential type name for JSON-sourced credentials, or the + // proto type URL for proto-sourced ones. + Type() string + // Equal reports whether other describes the same configuration. + Equal(other Identity) bool +} + +// jsonIdentity identifies a credential configured in the bootstrap file. +type jsonIdentity struct { + typ string + config json.RawMessage +} + +// NewJSONIdentity returns the Identity of a credential configured in the +// bootstrap file with the given type name and JSON configuration. +func NewJSONIdentity(typ string, config json.RawMessage) Identity { + return jsonIdentity{typ: typ, config: config} +} + +func (j jsonIdentity) Type() string { + return j.typ +} + +func (j jsonIdentity) Equal(other Identity) bool { + o, ok := other.(jsonIdentity) + return ok && j.typ == o.typ && bytes.Equal(j.config, o.config) +} + +// protoIdentity identifies a credential configured by a GrpcService proto +// credentials plugin. +type protoIdentity struct { + typeURL string + value []byte +} + +// NewProtoIdentity returns the Identity of a credential configured by the +// given GrpcService credentials plugin config. +func NewProtoIdentity(config *anypb.Any) Identity { + return protoIdentity{typeURL: config.GetTypeUrl(), value: config.GetValue()} +} + +func (p protoIdentity) Type() string { + return p.typeURL +} + +func (p protoIdentity) Equal(other Identity) bool { + o, ok := other.(protoIdentity) + return ok && p.typeURL == o.typeURL && bytes.Equal(p.value, o.value) +} + +// ChannelCreds pairs a built credentials bundle with the identity of the +// configuration it was built from. +type ChannelCreds struct { + bundle credentials.Bundle + identity Identity + cleanup func() +} + +// NewChannelCreds pairs the given bundle with its identity. cleanup releases +// the resources held by the bundle and is run by Close; it must be nil when +// the bundle is owned by another component (e.g. the bootstrap config), in +// which case Close is a no-op. +func NewChannelCreds(bundle credentials.Bundle, identity Identity, cleanup func()) *ChannelCreds { + if cleanup != nil { + cleanup = sync.OnceFunc(cleanup) + } + return &ChannelCreds{bundle: bundle, identity: identity, cleanup: cleanup} +} + +// Bundle returns the built credentials bundle. +func (c *ChannelCreds) Bundle() credentials.Bundle { + return c.bundle +} + +// Equal reports whether c and other were built from the same configuration. +func (c *ChannelCreds) Equal(other *ChannelCreds) bool { + if c == nil || other == nil { + return c == other + } + return c.identity.Equal(other.identity) +} + +// Close releases the resources held by the bundle, if owned. It is +// idempotent. +func (c *ChannelCreds) Close() { + if c != nil && c.cleanup != nil { + c.cleanup() + } +} + +// CallCreds pairs built per-RPC credentials with the identity of the +// configuration they were built from. +type CallCreds struct { + creds credentials.PerRPCCredentials + identity Identity + cleanup func() +} + +// NewCallCreds pairs the given per-RPC credentials with their identity. +// cleanup releases the resources held by the credentials and is run by Close; +// it must be nil when the credentials are owned by another component (e.g. +// the bootstrap config), in which case Close is a no-op. +func NewCallCreds(creds credentials.PerRPCCredentials, identity Identity, cleanup func()) *CallCreds { + if cleanup != nil { + cleanup = sync.OnceFunc(cleanup) + } + return &CallCreds{creds: creds, identity: identity, cleanup: cleanup} +} + +// Credentials returns the built per-RPC credentials. +func (c *CallCreds) Credentials() credentials.PerRPCCredentials { + return c.creds +} + +// Equal reports whether c and other were built from the same configuration. +func (c *CallCreds) Equal(other *CallCreds) bool { + if c == nil || other == nil { + return c == other + } + return c.identity.Equal(other.identity) +} + +// Close releases the resources held by the credentials, if owned. It is +// idempotent. +func (c *CallCreds) Close() { + if c != nil && c.cleanup != nil { + c.cleanup() + } +} diff --git a/internal/xds/grpcservice/credsregistry/access_token.go b/internal/xds/grpcservice/credsregistry/access_token.go new file mode 100644 index 000000000000..0722f4cc294a --- /dev/null +++ b/internal/xds/grpcservice/credsregistry/access_token.go @@ -0,0 +1,62 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package credsregistry + +import ( + "encoding/json" + "fmt" + + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/internal/xds/grpcservice/accesstokencreds" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/anypb" + + accesstokenpb "github.com/envoyproxy/go-control-plane/envoy/extensions/grpc_service/call_credentials/access_token/v3" +) + +const accessTokenCredsTypeURL = "type.googleapis.com/envoy.extensions.grpc_service.call_credentials.access_token.v3.AccessTokenCredentials" + +func init() { + RegisterCallCredsBuilder(accessTokenCredsTypeURL, accessTokenCredsBuilder{}) +} + +// accessTokenCredsBuilder builds static access token call credentials from an +// AccessTokenCredentials plugin config. +type accessTokenCredsBuilder struct{} + +func (accessTokenCredsBuilder) Build(config *anypb.Any) (credentials.PerRPCCredentials, func(), error) { + var accessToken accesstokenpb.AccessTokenCredentials + if err := anypb.UnmarshalTo(config, &accessToken, proto.UnmarshalOptions{}); err != nil { + return nil, nil, fmt.Errorf("failed to unmarshal AccessTokenCredentials: %v", err) + } + if accessToken.GetToken() == "" { + return nil, nil, fmt.Errorf("access token must be non-empty") + } + cfgJSON, err := json.Marshal(map[string]string{"token": accessToken.GetToken()}) + if err != nil { + return nil, nil, fmt.Errorf("failed to marshal access token config: %v", err) + } + cc, err := accesstokencreds.NewCallCredentials(cfgJSON) + if err != nil { + return nil, nil, err + } + // These credentials hold no resources; the no-op cleanup satisfies the + // registry's Build contract. + return cc, func() {}, nil +} diff --git a/internal/xds/grpcservice/credsregistry/credsregistry.go b/internal/xds/grpcservice/credsregistry/credsregistry.go new file mode 100644 index 000000000000..bb88e588940e --- /dev/null +++ b/internal/xds/grpcservice/credsregistry/credsregistry.go @@ -0,0 +1,80 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +// Package credsregistry contains registries of builders for the channel and +// call credentials that may be configured in a GrpcService proto, keyed by +// the proto type URL of their configuration (gRFC A102). +package credsregistry + +import ( + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/internal/xds/bootstrap" + "google.golang.org/protobuf/types/known/anypb" +) + +var ( + // channelCredsBuilders is a map from proto type URL to + // ChannelCredsBuilder. + channelCredsBuilders = make(map[string]ChannelCredsBuilder) + // callCredsBuilders is a map from proto type URL to CallCredsBuilder. + callCredsBuilders = make(map[string]CallCredsBuilder) +) + +// ChannelCredsBuilder builds channel credentials from a GrpcService channel +// credentials plugin config. +type ChannelCredsBuilder interface { + // Build creates a credentials bundle from the given plugin config. The + // bootstrap configuration is available to builders that reference + // resources configured there (e.g. certificate provider instances); + // builders that do not need it ignore it. The returned function releases + // the resources held by the bundle when it is no longer needed. + Build(config *anypb.Any, bc *bootstrap.Config) (credentials.Bundle, func(), error) +} + +// CallCredsBuilder builds call credentials from a GrpcService call +// credentials plugin config. +type CallCredsBuilder interface { + // Build creates per-RPC credentials from the given plugin config. The + // returned function releases the resources held by the credentials when + // they are no longer needed. + Build(config *anypb.Any) (credentials.PerRPCCredentials, func(), error) +} + +// RegisterChannelCredsBuilder registers the builder for the given proto type +// URL. Must be called at init time. Not thread safe. +func RegisterChannelCredsBuilder(typeURL string, b ChannelCredsBuilder) { + channelCredsBuilders[typeURL] = b +} + +// GetChannelCredsBuilder returns the builder registered for the given proto +// type URL, or nil if there is none. +func GetChannelCredsBuilder(typeURL string) ChannelCredsBuilder { + return channelCredsBuilders[typeURL] +} + +// RegisterCallCredsBuilder registers the builder for the given proto type +// URL. Must be called at init time. Not thread safe. +func RegisterCallCredsBuilder(typeURL string, b CallCredsBuilder) { + callCredsBuilders[typeURL] = b +} + +// GetCallCredsBuilder returns the builder registered for the given proto type +// URL, or nil if there is none. +func GetCallCredsBuilder(typeURL string) CallCredsBuilder { + return callCredsBuilders[typeURL] +} diff --git a/internal/xds/grpcservice/credsregistry/google_default.go b/internal/xds/grpcservice/credsregistry/google_default.go new file mode 100644 index 000000000000..0775d68fe0b1 --- /dev/null +++ b/internal/xds/grpcservice/credsregistry/google_default.go @@ -0,0 +1,40 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package credsregistry + +import ( + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/credentials/google" + "google.golang.org/grpc/internal/xds/bootstrap" + "google.golang.org/protobuf/types/known/anypb" +) + +const googleDefaultCredsTypeURL = "type.googleapis.com/envoy.extensions.grpc_service.channel_credentials.google_default.v3.GoogleDefaultCredentials" + +func init() { + RegisterChannelCredsBuilder(googleDefaultCredsTypeURL, googleDefaultCredsBuilder{}) +} + +// googleDefaultCredsBuilder builds Google Default channel credentials from a +// GoogleDefaultCredentials plugin config. +type googleDefaultCredsBuilder struct{} + +func (googleDefaultCredsBuilder) Build(*anypb.Any, *bootstrap.Config) (credentials.Bundle, func(), error) { + return google.NewDefaultCredentials(), func() {}, nil +} diff --git a/internal/xds/grpcservice/credsregistry/insecure.go b/internal/xds/grpcservice/credsregistry/insecure.go new file mode 100644 index 000000000000..0cc44fe8f85d --- /dev/null +++ b/internal/xds/grpcservice/credsregistry/insecure.go @@ -0,0 +1,40 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package credsregistry + +import ( + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/internal/xds/bootstrap" + "google.golang.org/protobuf/types/known/anypb" +) + +const insecureCredsTypeURL = "type.googleapis.com/envoy.extensions.grpc_service.channel_credentials.insecure.v3.InsecureCredentials" + +func init() { + RegisterChannelCredsBuilder(insecureCredsTypeURL, insecureCredsBuilder{}) +} + +// insecureCredsBuilder builds insecure channel credentials from an +// InsecureCredentials plugin config. +type insecureCredsBuilder struct{} + +func (insecureCredsBuilder) Build(*anypb.Any, *bootstrap.Config) (credentials.Bundle, func(), error) { + return insecure.NewBundle(), func() {}, nil +} diff --git a/internal/xds/grpcservice/credsregistry/tls.go b/internal/xds/grpcservice/credsregistry/tls.go new file mode 100644 index 000000000000..ea4eae308dc8 --- /dev/null +++ b/internal/xds/grpcservice/credsregistry/tls.go @@ -0,0 +1,220 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package credsregistry + +import ( + "context" + "crypto/tls" + "errors" + "fmt" + "net" + "sync" + + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/credentials/tls/certprovider" + "google.golang.org/grpc/internal/xds/bootstrap" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/anypb" + + tlscredspb "github.com/envoyproxy/go-control-plane/envoy/extensions/grpc_service/channel_credentials/tls/v3" + v3tlspb "github.com/envoyproxy/go-control-plane/envoy/extensions/transport_sockets/tls/v3" +) + +const tlsCredsTypeURL = "type.googleapis.com/envoy.extensions.grpc_service.channel_credentials.tls.v3.TlsCredentials" + +func init() { + RegisterChannelCredsBuilder(tlsCredsTypeURL, tlsCredsBuilder{}) +} + +// tlsCredsBuilder builds TLS channel credentials from a TlsCredentials plugin +// config, whose root and identity certificates are sourced from certificate +// provider instances configured in the bootstrap config. +type tlsCredsBuilder struct{} + +func (tlsCredsBuilder) Build(config *anypb.Any, bc *bootstrap.Config) (credentials.Bundle, func(), error) { + var tlsCfg tlscredspb.TlsCredentials + if err := anypb.UnmarshalTo(config, &tlsCfg, proto.UnmarshalOptions{}); err != nil { + return nil, nil, fmt.Errorf("failed to unmarshal TlsCredentials: %v", err) + } + + // The certificate provider instance names are validated against the + // bootstrap config here, at parse time, the same way CommonTlsContext + // instances are (gRFC A29): an unknown instance name is a NACK. The + // providers themselves are instantiated lazily, on the first handshake, + // so that a parsed-but-never-dialed config does not start certificate + // watchers. + root := tlsCfg.GetRootCertificateProvider() + if root.GetInstanceName() == "" { + return nil, nil, fmt.Errorf("tls credentials must specify root_certificate_provider with an instance_name") + } + rootCfg, err := certProviderConfig(bc, root) + if err != nil { + return nil, nil, fmt.Errorf("tls credentials root certificate provider: %v", err) + } + creds := &providerTLSCreds{ + rootConfig: rootCfg, + rootCertName: root.GetCertificateName(), + } + if identity := tlsCfg.GetIdentityCertificateProvider(); identity != nil { + if identity.GetInstanceName() == "" { + return nil, nil, fmt.Errorf("tls credentials identity_certificate_provider must specify an instance_name") + } + identityCfg, err := certProviderConfig(bc, identity) + if err != nil { + return nil, nil, fmt.Errorf("tls credentials identity certificate provider: %v", err) + } + creds.identityConfig = identityCfg + creds.identityCertName = identity.GetCertificateName() + } + return &tlsBundle{creds: creds}, creds.close, nil +} + +// certProviderConfig looks up the certificate provider instance referenced by +// the given proto in the bootstrap config. +func certProviderConfig(bc *bootstrap.Config, instance *v3tlspb.CommonTlsContext_CertificateProviderInstance) (*certprovider.BuildableConfig, error) { + if bc == nil { + return nil, fmt.Errorf("no bootstrap configuration available to resolve certificate provider instances") + } + cfg, ok := bc.CertProviderConfigs()[instance.GetInstanceName()] + if !ok { + return nil, fmt.Errorf("certificate provider instance name %q missing in bootstrap configuration", instance.GetInstanceName()) + } + return cfg, nil +} + +// tlsBundle is a credentials.Bundle wrapping provider-backed TLS transport +// credentials. It carries no per-RPC credentials. +type tlsBundle struct { + creds *providerTLSCreds +} + +func (b *tlsBundle) TransportCredentials() credentials.TransportCredentials { + return b.creds +} + +func (b *tlsBundle) PerRPCCredentials() credentials.PerRPCCredentials { + return nil +} + +func (b *tlsBundle) NewWithMode(string) (credentials.Bundle, error) { + return nil, fmt.Errorf("xDS TLS channel credentials only support one mode") +} + +// providerTLSCreds is a client-side credentials.TransportCredentials that +// sources the server root CA certificates, and optionally the client identity +// certificates, from certificate provider instances. The key material is +// fetched from the providers on every handshake, so certificate reloads are +// picked up; the providers themselves are instantiated on the first +// handshake. +type providerTLSCreds struct { + rootConfig *certprovider.BuildableConfig + rootCertName string + identityConfig *certprovider.BuildableConfig // nil when no identity certificate is configured + identityCertName string + + mu sync.Mutex + closed bool + rootProvider certprovider.Provider + identityProvider certprovider.Provider +} + +// providers instantiates the certificate providers on first use and returns +// them. It fails if the credentials have already been closed. +func (c *providerTLSCreds) providers() (root, identity certprovider.Provider, err error) { + c.mu.Lock() + defer c.mu.Unlock() + if c.closed { + return nil, nil, errors.New("xDS TLS channel credentials have been closed") + } + if c.rootProvider == nil { + p, err := c.rootConfig.Build(certprovider.BuildOptions{CertName: c.rootCertName, WantRoot: true}) + if err != nil { + return nil, nil, fmt.Errorf("failed to build root certificate provider: %v", err) + } + c.rootProvider = p + } + if c.identityConfig != nil && c.identityProvider == nil { + p, err := c.identityConfig.Build(certprovider.BuildOptions{CertName: c.identityCertName, WantIdentity: true}) + if err != nil { + return nil, nil, fmt.Errorf("failed to build identity certificate provider: %v", err) + } + c.identityProvider = p + } + return c.rootProvider, c.identityProvider, nil +} + +// close releases the certificate providers, if they were instantiated. +// Subsequent handshakes fail. +func (c *providerTLSCreds) close() { + c.mu.Lock() + defer c.mu.Unlock() + c.closed = true + if c.rootProvider != nil { + c.rootProvider.Close() + c.rootProvider = nil + } + if c.identityProvider != nil { + c.identityProvider.Close() + c.identityProvider = nil + } +} + +func (c *providerTLSCreds) ClientHandshake(ctx context.Context, authority string, rawConn net.Conn) (net.Conn, credentials.AuthInfo, error) { + root, identity, err := c.providers() + if err != nil { + return nil, nil, err + } + rootKM, err := root.KeyMaterial(ctx) + if err != nil { + return nil, nil, fmt.Errorf("failed to get root certificates: %v", err) + } + if rootKM.Roots == nil { + return nil, nil, errors.New("root certificate provider returned no root certificates") + } + cfg := &tls.Config{RootCAs: rootKM.Roots} + if identity != nil { + identityKM, err := identity.KeyMaterial(ctx) + if err != nil { + return nil, nil, fmt.Errorf("failed to get identity certificates: %v", err) + } + cfg.Certificates = identityKM.Certs + } + return credentials.NewTLS(cfg).ClientHandshake(ctx, authority, rawConn) +} + +func (c *providerTLSCreds) ServerHandshake(net.Conn) (net.Conn, credentials.AuthInfo, error) { + return nil, nil, errors.New("server handshake is not supported by xDS TLS channel credentials") +} + +func (c *providerTLSCreds) Info() credentials.ProtocolInfo { + return credentials.ProtocolInfo{SecurityProtocol: "tls"} +} + +func (c *providerTLSCreds) Clone() credentials.TransportCredentials { + return &providerTLSCreds{ + rootConfig: c.rootConfig, + rootCertName: c.rootCertName, + identityConfig: c.identityConfig, + identityCertName: c.identityCertName, + } +} + +func (c *providerTLSCreds) OverrideServerName(string) error { + return errors.New("overriding server name is not supported by xDS TLS channel credentials") +} diff --git a/internal/xds/grpcservice/credsregistry/tls_test.go b/internal/xds/grpcservice/credsregistry/tls_test.go new file mode 100644 index 000000000000..ed4daef446dd --- /dev/null +++ b/internal/xds/grpcservice/credsregistry/tls_test.go @@ -0,0 +1,287 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package credsregistry + +import ( + "context" + "crypto/tls" + "crypto/x509" + "encoding/json" + "fmt" + "net" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "google.golang.org/grpc/internal/grpctest" + "google.golang.org/grpc/internal/xds/bootstrap" + "google.golang.org/grpc/testdata" + "google.golang.org/protobuf/types/known/anypb" + + tlscredspb "github.com/envoyproxy/go-control-plane/envoy/extensions/grpc_service/channel_credentials/tls/v3" + v3tlspb "github.com/envoyproxy/go-control-plane/envoy/extensions/transport_sockets/tls/v3" +) + +type s struct { + grpctest.Tester +} + +func Test(t *testing.T) { + grpctest.RunSubTests(t, s{}) +} + +const defaultTestTimeout = 10 * time.Second + +// testBootstrapConfig returns a bootstrap config with two certificate +// provider instances: "root-instance" watching the CA certificate that signed +// the test server's certificate, and "identity-instance" watching a client +// certificate and key. +func testBootstrapConfig(t *testing.T) *bootstrap.Config { + t.Helper() + + // Use forward slashes so the paths survive JSON encoding on Windows. + rootCert := filepath.ToSlash(testdata.Path("x509/server_ca_cert.pem")) + clientCert := filepath.ToSlash(testdata.Path("x509/client1_cert.pem")) + clientKey := filepath.ToSlash(testdata.Path("x509/client1_key.pem")) + + contents, err := bootstrap.NewContentsForTesting(bootstrap.ConfigOptionsForTesting{ + Servers: json.RawMessage(`[{"server_uri": "passthrough:///unused", "channel_creds": [{"type": "insecure"}]}]`), + Node: json.RawMessage(`{"id": "test-node"}`), + CertificateProviders: map[string]json.RawMessage{ + "root-instance": json.RawMessage(fmt.Sprintf(`{ + "plugin_name": "file_watcher", + "config": {"ca_certificate_file": %q, "refresh_interval": "600s"} + }`, rootCert)), + "identity-instance": json.RawMessage(fmt.Sprintf(`{ + "plugin_name": "file_watcher", + "config": {"certificate_file": %q, "private_key_file": %q, "refresh_interval": "600s"} + }`, clientCert, clientKey)), + }, + }) + if err != nil { + t.Fatalf("NewContentsForTesting() failed: %v", err) + } + cfg, err := bootstrap.NewConfigFromContents(contents) + if err != nil { + t.Fatalf("NewConfigFromContents() failed: %v", err) + } + return cfg +} + +// tlsCredsConfig returns a marshaled TlsCredentials plugin config referencing +// the given provider instance names. An empty identity omits the identity +// certificate provider. +func tlsCredsConfig(t *testing.T, root, identity string) *anypb.Any { + t.Helper() + cfg := &tlscredspb.TlsCredentials{} + if root != "" { + cfg.RootCertificateProvider = &v3tlspb.CommonTlsContext_CertificateProviderInstance{InstanceName: root} + } + if identity != "" { + cfg.IdentityCertificateProvider = &v3tlspb.CommonTlsContext_CertificateProviderInstance{InstanceName: identity} + } + a, err := anypb.New(cfg) + if err != nil { + t.Fatalf("Failed to marshal TlsCredentials: %v", err) + } + return a +} + +// Tests that building TLS channel credentials validates the certificate +// provider instance names against the bootstrap config. +func (s) TestTLSCredsBuild_Errors(t *testing.T) { + bc := testBootstrapConfig(t) + + // The tlsCredsConfig helper cannot express an identity certificate + // provider with an empty instance name, so build that config directly. + emptyIdentityInstance, err := anypb.New(&tlscredspb.TlsCredentials{ + RootCertificateProvider: &v3tlspb.CommonTlsContext_CertificateProviderInstance{InstanceName: "root-instance"}, + IdentityCertificateProvider: &v3tlspb.CommonTlsContext_CertificateProviderInstance{}, + }) + if err != nil { + t.Fatalf("Failed to marshal TlsCredentials: %v", err) + } + + tests := []struct { + name string + config *anypb.Any + bc *bootstrap.Config + wantErr string + }{ + { + name: "unmarshal_failure", + config: &anypb.Any{TypeUrl: tlsCredsTypeURL, Value: []byte{0xff}}, + bc: bc, + wantErr: "failed to unmarshal TlsCredentials", + }, + { + name: "missing_root_certificate_provider", + config: tlsCredsConfig(t, "", "identity-instance"), + bc: bc, + wantErr: "must specify root_certificate_provider", + }, + { + name: "empty_identity_instance_name", + config: emptyIdentityInstance, + bc: bc, + wantErr: "identity_certificate_provider must specify an instance_name", + }, + { + name: "unknown_root_instance", + config: tlsCredsConfig(t, "unknown-instance", ""), + bc: bc, + wantErr: `"unknown-instance" missing in bootstrap`, + }, + { + name: "unknown_identity_instance", + config: tlsCredsConfig(t, "root-instance", "unknown-instance"), + bc: bc, + wantErr: `"unknown-instance" missing in bootstrap`, + }, + { + name: "nil_bootstrap_config", + config: tlsCredsConfig(t, "root-instance", ""), + bc: nil, + wantErr: "no bootstrap configuration", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, _, err := GetChannelCredsBuilder(tlsCredsTypeURL).Build(tt.config, tt.bc) + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("Build() returned error %v, want error containing %q", err, tt.wantErr) + } + }) + } +} + +// startTestTLSServer starts a TLS server that performs one handshake per +// accepted connection. If mTLS is true, the server requires and verifies a +// client certificate. +func startTestTLSServer(t *testing.T, mTLS bool) net.Listener { + t.Helper() + + serverCert, err := tls.LoadX509KeyPair(testdata.Path("x509/server1_cert.pem"), testdata.Path("x509/server1_key.pem")) + if err != nil { + t.Fatalf("Failed to load server certificate: %v", err) + } + // gRPC's TLS credentials enforce ALPN, so the server must advertise h2. + cfg := &tls.Config{Certificates: []tls.Certificate{serverCert}, NextProtos: []string{"h2"}} + if mTLS { + pem, err := os.ReadFile(testdata.Path("x509/client_ca_cert.pem")) + if err != nil { + t.Fatalf("Failed to read client CA certificate: %v", err) + } + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(pem) { + t.Fatal("Failed to parse client CA certificate") + } + cfg.ClientAuth = tls.RequireAndVerifyClientCert + cfg.ClientCAs = pool + } + + lis, err := tls.Listen("tcp", "localhost:0", cfg) + if err != nil { + t.Fatalf("Failed to start test TLS server: %v", err) + } + t.Cleanup(func() { lis.Close() }) + go func() { + for { + conn, err := lis.Accept() + if err != nil { + return + } + go func() { + conn.(*tls.Conn).Handshake() + conn.Close() + }() + } + }() + return lis +} + +// clientHandshake dials the test server and performs a client-side TLS +// handshake with credentials built from the given plugin config. +func clientHandshake(t *testing.T, config *anypb.Any, bc *bootstrap.Config) error { + t.Helper() + + bundle, cleanup, err := GetChannelCredsBuilder(tlsCredsTypeURL).Build(config, bc) + if err != nil { + t.Fatalf("Build() failed: %v", err) + } + defer cleanup() + + mTLS := false + var cfg tlscredspb.TlsCredentials + if err := config.UnmarshalTo(&cfg); err == nil && cfg.GetIdentityCertificateProvider() != nil { + mTLS = true + } + lis := startTestTLSServer(t, mTLS) + + rawConn, err := net.Dial("tcp", lis.Addr().String()) + if err != nil { + t.Fatalf("Failed to dial test server: %v", err) + } + defer rawConn.Close() + + ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) + defer cancel() + // The test server certificate is issued for *.test.example.com. + _, _, err = bundle.TransportCredentials().ClientHandshake(ctx, "x.test.example.com", rawConn) + return err +} + +// Tests that TLS channel credentials backed by certificate provider instances +// complete a TLS handshake, with and without an identity certificate. +func (s) TestTLSCredsHandshake(t *testing.T) { + bc := testBootstrapConfig(t) + + if err := clientHandshake(t, tlsCredsConfig(t, "root-instance", ""), bc); err != nil { + t.Fatalf("ClientHandshake() with root-only TLS credentials failed: %v", err) + } + if err := clientHandshake(t, tlsCredsConfig(t, "root-instance", "identity-instance"), bc); err != nil { + t.Fatalf("ClientHandshake() with mTLS credentials failed: %v", err) + } +} + +// Tests that closed TLS channel credentials fail handshakes, and that closing +// credentials that never performed a handshake is safe. +func (s) TestTLSCredsClose(t *testing.T) { + bc := testBootstrapConfig(t) + + // Closing credentials whose providers were never instantiated must be a + // no-op. + bundle, cleanup, err := GetChannelCredsBuilder(tlsCredsTypeURL).Build(tlsCredsConfig(t, "root-instance", ""), bc) + if err != nil { + t.Fatalf("Build() failed: %v", err) + } + cleanup() + + // A handshake after close must fail. + server, client := net.Pipe() + defer server.Close() + defer client.Close() + ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) + defer cancel() + if _, _, err := bundle.TransportCredentials().ClientHandshake(ctx, "x.test.example.com", client); err == nil || !strings.Contains(err.Error(), "closed") { + t.Fatalf("ClientHandshake() after close returned error %v, want error containing %q", err, "closed") + } +} diff --git a/internal/xds/grpcservice/credsregistry/xds.go b/internal/xds/grpcservice/credsregistry/xds.go new file mode 100644 index 000000000000..034b94ea1e29 --- /dev/null +++ b/internal/xds/grpcservice/credsregistry/xds.go @@ -0,0 +1,58 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package credsregistry + +import ( + "fmt" + + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/internal/xds/bootstrap" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/anypb" + + xdscredspb "github.com/envoyproxy/go-control-plane/envoy/extensions/grpc_service/channel_credentials/xds/v3" +) + +const xdsCredsTypeURL = "type.googleapis.com/envoy.extensions.grpc_service.channel_credentials.xds.v3.XdsCredentials" + +func init() { + RegisterChannelCredsBuilder(xdsCredsTypeURL, xdsCredsBuilder{}) +} + +// xdsCredsBuilder builds channel credentials from an XdsCredentials plugin +// config. A side-channel target is not an xDS cluster, so there is no xDS +// security configuration for it; the xds credential therefore resolves to its +// required fallback credential, whose builder is looked up in the registry. +type xdsCredsBuilder struct{} + +func (xdsCredsBuilder) Build(config *anypb.Any, bc *bootstrap.Config) (credentials.Bundle, func(), error) { + var xdsCfg xdscredspb.XdsCredentials + if err := anypb.UnmarshalTo(config, &xdsCfg, proto.UnmarshalOptions{}); err != nil { + return nil, nil, fmt.Errorf("failed to unmarshal XdsCredentials: %v", err) + } + fallback := xdsCfg.GetFallbackCredentials() + if fallback == nil { + return nil, nil, fmt.Errorf("xds credentials missing required fallback credentials") + } + b := GetChannelCredsBuilder(fallback.GetTypeUrl()) + if b == nil { + return nil, nil, fmt.Errorf("unsupported fallback credentials type %q in xds credentials", fallback.GetTypeUrl()) + } + return b.Build(fallback, bc) +} diff --git a/internal/xds/grpcservice/credsregistry/xds_test.go b/internal/xds/grpcservice/credsregistry/xds_test.go new file mode 100644 index 000000000000..996850d076de --- /dev/null +++ b/internal/xds/grpcservice/credsregistry/xds_test.go @@ -0,0 +1,73 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package credsregistry + +import ( + "strings" + "testing" + + "google.golang.org/protobuf/types/known/anypb" + + xdscredspb "github.com/envoyproxy/go-control-plane/envoy/extensions/grpc_service/channel_credentials/xds/v3" +) + +// Tests that building xds channel credentials fails on malformed configs and +// on missing or unsupported fallback credentials. +func (s) TestXDSCredsBuild_Errors(t *testing.T) { + missingFallback, err := anypb.New(&xdscredspb.XdsCredentials{}) + if err != nil { + t.Fatalf("Failed to marshal XdsCredentials: %v", err) + } + unsupportedFallback, err := anypb.New(&xdscredspb.XdsCredentials{ + FallbackCredentials: &anypb.Any{TypeUrl: "type.googleapis.com/unknown.Credentials"}, + }) + if err != nil { + t.Fatalf("Failed to marshal XdsCredentials: %v", err) + } + + tests := []struct { + name string + config *anypb.Any + wantErr string + }{ + { + name: "unmarshal_failure", + config: &anypb.Any{TypeUrl: xdsCredsTypeURL, Value: []byte{0xff}}, + wantErr: "failed to unmarshal XdsCredentials", + }, + { + name: "missing_fallback_credentials", + config: missingFallback, + wantErr: "missing required fallback credentials", + }, + { + name: "unsupported_fallback_credentials_type", + config: unsupportedFallback, + wantErr: "unsupported fallback credentials type", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, _, err := GetChannelCredsBuilder(xdsCredsTypeURL).Build(tt.config, nil) + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("Build() returned error %v, want error containing %q", err, tt.wantErr) + } + }) + } +} diff --git a/internal/xds/grpcservice/grpcservice.go b/internal/xds/grpcservice/grpcservice.go new file mode 100644 index 000000000000..da170a55cd8f --- /dev/null +++ b/internal/xds/grpcservice/grpcservice.go @@ -0,0 +1,296 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +// Package grpcservice parses and validates envoy GrpcService protos into a +// form usable for creating side-channel gRPC connections. +package grpcservice + +import ( + "fmt" + "net/url" + "slices" + "strings" + "time" + + imetadata "google.golang.org/grpc/internal/metadata" + "google.golang.org/grpc/internal/xds/bootstrap" + "google.golang.org/grpc/internal/xds/grpcservice/creds" + "google.golang.org/grpc/internal/xds/grpcservice/credsregistry" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/resolver" + + v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" + "google.golang.org/protobuf/types/known/anypb" +) + +const ( + maxHeaderKeyLen = 16384 + maxHeaderValueLen = 16384 +) + +// Config is the parsed form of a GrpcService proto. It carries the built, +// ready-to-use credentials for the side channel. +type Config struct { + // TargetURI is the gRPC target URI of the side-channel service. + TargetURI string + // Timeout, if non-zero, is the deadline to use for RPCs on the side + // channel. + Timeout time.Duration + // InitialMetadata is the metadata to add to RPCs on the side channel. + InitialMetadata metadata.MD + // ChannelCredentials are the channel credentials to create the side + // channel with, paired with the identity of their source configuration. + ChannelCredentials *creds.ChannelCreds + // CallCredentials are the call credentials to apply to RPCs sent on the + // side channel, paired with the identities of their source + // configurations, preserving order. + CallCredentials []*creds.CallCreds +} + +// Equal reports whether c and other describe the same side channel: the same +// target with the same channel and call credential identities. Timeout and +// initial metadata are applied per-RPC and intentionally do not affect +// channel sharing. +func (c *Config) Equal(other *Config) bool { + if c == nil || other == nil { + return c == other + } + return c.TargetURI == other.TargetURI && + c.ChannelCredentials.Equal(other.ChannelCredentials) && + slices.EqualFunc(c.CallCredentials, other.CallCredentials, (*creds.CallCreds).Equal) +} + +// Close releases the credentials owned by the config. It is idempotent, and a +// no-op for credentials owned by another component (e.g. the allowlisted +// credentials owned by the bootstrap config). +func (c *Config) Close() { + if c == nil { + return + } + c.ChannelCredentials.Close() + for _, cc := range c.CallCredentials { + cc.Close() + } +} + +// Parse parses and validates a GrpcService proto into a Config, applying the +// gRFC A102 trust policy. +// +// Credentials configured in the proto are honored only when the xDS +// management server that delivered it is trusted, i.e. configured with the +// trusted_xds_server server feature; a nil server config means the delivering +// management server is unknown and is treated as untrusted. For untrusted +// management servers the target must be present in the bootstrap +// allowed_grpc_services map, and the returned Config carries the credentials +// configured there. +// +// The credentials in the returned Config are built and ready to use. Owned +// credentials are released by Config.Close; the xDS client's CreateChannel +// takes over that responsibility when a channel is created from the Config. +func Parse(gs *v3corepb.GrpcService, bc *bootstrap.Config, sc *bootstrap.ServerConfig) (_ *Config, err error) { + googleGrpc := gs.GetGoogleGrpc() + if googleGrpc == nil { + return nil, fmt.Errorf("grpcservice: only google_grpc GrpcService config is supported") + } + + targetURI := googleGrpc.GetTargetUri() + if targetURI == "" { + return nil, fmt.Errorf("grpcservice: target_uri must be non-empty") + } + if err := validateTargetURI(targetURI); err != nil { + return nil, err + } + + cfg := &Config{TargetURI: targetURI} + // Release any credentials built before a mid-parse failure. + defer func() { + if err != nil { + cfg.Close() + } + }() + + if trusted := sc != nil && sc.ServerFeaturesTrustedXDSServer(); trusted { + if cfg.ChannelCredentials, err = buildChannelCredentials(googleGrpc.GetChannelCredentialsPlugin(), bc); err != nil { + return nil, fmt.Errorf("grpcservice: %v", err) + } + if cfg.CallCredentials, err = buildCallCredentials(googleGrpc.GetCallCredentialsPlugin()); err != nil { + return nil, fmt.Errorf("grpcservice: %v", err) + } + } else { + // A nil bootstrap config has no allowlist, so all targets are + // rejected. + var svc *bootstrap.AllowedGRPCService + if bc != nil { + svc = bc.AllowedGRPCService(targetURI) + } + if svc == nil { + return nil, fmt.Errorf("grpcservice: target_uri %q is not present in allowed_grpc_services", targetURI) + } + // The allowlisted credentials are owned by the bootstrap config: + // their pairs carry no cleanup, so Config.Close does not affect + // them. + cfg.ChannelCredentials, cfg.CallCredentials = svc.SideChannelCredentials() + } + + if cfg.Timeout, err = parseTimeout(gs); err != nil { + return nil, err + } + if cfg.InitialMetadata, err = parseInitialMetadata(gs.GetInitialMetadata()); err != nil { + return nil, err + } + return cfg, nil +} + +// buildChannelCredentials builds the first channel credential from the plugin +// list whose proto type has a registered builder. It is an error if none of +// the configured plugins are supported, or if building the selected plugin +// fails. +func buildChannelCredentials(plugins []*anypb.Any, bc *bootstrap.Config) (*creds.ChannelCreds, error) { + for _, p := range plugins { + if p == nil { + continue + } + b := credsregistry.GetChannelCredsBuilder(p.GetTypeUrl()) + if b == nil { + continue + } + bundle, cleanup, err := b.Build(p, bc) + if err != nil { + return nil, fmt.Errorf("failed to build channel credentials %q: %v", p.GetTypeUrl(), err) + } + return creds.NewChannelCreds(bundle, creds.NewProtoIdentity(p), cleanup), nil + } + return nil, fmt.Errorf("no supported channel credentials found in grpc_service") +} + +// buildCallCredentials builds the call credentials from the plugin list, +// preserving order. Plugins whose proto type has no registered builder are +// skipped; call credentials are optional, so an empty result is not an error. +func buildCallCredentials(plugins []*anypb.Any) (_ []*creds.CallCreds, err error) { + var out []*creds.CallCreds + // Release any credentials built before a mid-iteration failure. + defer func() { + if err != nil { + for _, cc := range out { + cc.Close() + } + } + }() + for _, p := range plugins { + if p == nil { + continue + } + b := credsregistry.GetCallCredsBuilder(p.GetTypeUrl()) + if b == nil { + continue + } + cc, cleanup, err := b.Build(p) + if err != nil { + return nil, fmt.Errorf("failed to build call credentials %q: %v", p.GetTypeUrl(), err) + } + out = append(out, creds.NewCallCreds(cc, creds.NewProtoIdentity(p), cleanup)) + } + return out, nil +} + +// validateTargetURI verifies that the target URI can be handled by a +// registered resolver. +func validateTargetURI(targetURI string) error { + // Mirror the scheme resolution performed by grpc.NewClient: use the + // target's scheme if it parses and is registered; otherwise fall back + // to the default scheme with the whole target as the endpoint. + if u, err := url.Parse(targetURI); err == nil && resolver.Get(u.Scheme) != nil { + return nil + } + canonicalTarget := resolver.GetDefaultScheme() + ":///" + targetURI + u, err := url.Parse(canonicalTarget) + if err != nil { + return fmt.Errorf("grpcservice: target_uri %q is invalid: %v", targetURI, err) + } + if resolver.Get(u.Scheme) == nil { + return fmt.Errorf("grpcservice: no resolver for default scheme %q", u.Scheme) + } + return nil +} + +// parseTimeout validates and converts the GrpcService timeout. A zero timeout +// (unset) is allowed; any set value must be strictly positive. +func parseTimeout(gs *v3corepb.GrpcService) (time.Duration, error) { + d := gs.GetTimeout() + if d == nil { + return 0, nil + } + if err := d.CheckValid(); err != nil { + return 0, fmt.Errorf("grpcservice: invalid timeout: %v", err) + } + timeout := d.AsDuration() + if timeout <= 0 { + return 0, fmt.Errorf("grpcservice: timeout must be strictly positive, got %v", timeout) + } + return timeout, nil +} + +// parseInitialMetadata validates the HeaderValue protos and returns them as +// metadata.MD, preserving the control-plane order for each key. +func parseInitialMetadata(headers []*v3corepb.HeaderValue) (metadata.MD, error) { + if len(headers) == 0 { + return nil, nil + } + md := metadata.MD{} + for _, h := range headers { + key := h.GetKey() + // raw_value takes precedence over the legacy value field. + val := h.GetValue() + if len(h.GetRawValue()) > 0 { + val = string(h.GetRawValue()) + } + if err := validateHeaderKey(key); err != nil { + return nil, fmt.Errorf("grpcservice: invalid header key %q: %v", key, err) + } + if err := validateHeaderValue(key, val); err != nil { + return nil, fmt.Errorf("grpcservice: invalid value for header key %q: %v", key, err) + } + md.Append(key, val) + } + return md, nil +} + +func validateHeaderKey(key string) error { + switch { + case len(key) > maxHeaderKeyLen: + return fmt.Errorf("header key exceeds maximum allowed length of %d", maxHeaderKeyLen) + case key == "host": + return fmt.Errorf("header key cannot be %q", "host") + case strings.HasPrefix(key, ":"): + // imetadata.ValidateKey ignores pseudo-headers, but gRFC A102 + // requires them to be rejected. + return fmt.Errorf("header key cannot start with %q", ":") + case strings.HasPrefix(key, "grpc-"): + return fmt.Errorf("header key cannot start with %q", "grpc-") + } + return imetadata.ValidateKey(key) +} + +func validateHeaderValue(key, val string) error { + if len(val) > maxHeaderValueLen { + return fmt.Errorf("header value exceeds maximum allowed length of %d", maxHeaderValueLen) + } + // ValidatePair skips value validation for "-bin" keys, which may carry + // arbitrary bytes. + return imetadata.ValidatePair(key, val) +} diff --git a/internal/xds/grpcservice/grpcservice_test.go b/internal/xds/grpcservice/grpcservice_test.go new file mode 100644 index 000000000000..194722f45151 --- /dev/null +++ b/internal/xds/grpcservice/grpcservice_test.go @@ -0,0 +1,316 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package grpcservice + +import ( + "encoding/json" + "strings" + "testing" + + "github.com/google/go-cmp/cmp" + "google.golang.org/grpc/internal/envconfig" + "google.golang.org/grpc/internal/grpctest" + "google.golang.org/grpc/internal/testutils" + "google.golang.org/grpc/internal/xds/bootstrap" + "google.golang.org/grpc/internal/xds/grpcservice/creds" + "google.golang.org/grpc/metadata" + + v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" + accesstokenpb "github.com/envoyproxy/go-control-plane/envoy/extensions/grpc_service/call_credentials/access_token/v3" + xdscredspb "github.com/envoyproxy/go-control-plane/envoy/extensions/grpc_service/channel_credentials/xds/v3" + "google.golang.org/protobuf/types/known/anypb" + "google.golang.org/protobuf/types/known/durationpb" +) + +type s struct { + grpctest.Tester +} + +func Test(t *testing.T) { + grpctest.RunSubTests(t, s{}) +} + +const ( + target = "dns:///my-service:443" + + insecureCredsTypeURL = "type.googleapis.com/envoy.extensions.grpc_service.channel_credentials.insecure.v3.InsecureCredentials" +) + +// bootstrapConfig builds a bootstrap Config whose allowed_grpc_services is set +// to the provided JSON (a map from target URI to allowed service config). +func bootstrapConfig(t *testing.T, allowed string) *bootstrap.Config { + t.Helper() + // The allowed_grpc_services bootstrap field is parsed only when a + // consuming feature is enabled. + testutils.SetEnvConfig(t, &envconfig.XDSClientExtProcEnabled, true) + contents, err := bootstrap.NewContentsForTesting(bootstrap.ConfigOptionsForTesting{ + Servers: json.RawMessage(`[{"server_uri":"td.googleapis.com:443","channel_creds":[{"type":"insecure"}]}]`), + Node: json.RawMessage(`{}`), + AllowedGRPCServices: json.RawMessage(allowed), + }) + if err != nil { + t.Fatalf("NewContentsForTesting() failed: %v", err) + } + cfg, err := bootstrap.NewConfigFromContents(contents) + if err != nil { + t.Fatalf("NewConfigFromContents() failed: %v", err) + } + return cfg +} + +// trustedServerConfig returns a server config carrying the trusted_xds_server +// server feature. +func trustedServerConfig(t *testing.T) *bootstrap.ServerConfig { + t.Helper() + sc, err := bootstrap.ServerConfigForTesting(bootstrap.ServerConfigTestingOptions{ + URI: "trusted-server:443", + ServerFeatures: []string{"trusted_xds_server"}, + }) + if err != nil { + t.Fatalf("ServerConfigForTesting() failed: %v", err) + } + return sc +} + +func googleGrpcService(target string, channelPlugins []*anypb.Any, timeout *durationpb.Duration) *v3corepb.GrpcService { + return &v3corepb.GrpcService{ + TargetSpecifier: &v3corepb.GrpcService_GoogleGrpc_{ + GoogleGrpc: &v3corepb.GrpcService_GoogleGrpc{ + TargetUri: target, + ChannelCredentialsPlugin: channelPlugins, + }, + }, + Timeout: timeout, + } +} + +func accessTokenPlugin(t *testing.T, token string) *anypb.Any { + t.Helper() + a, err := anypb.New(&accesstokenpb.AccessTokenCredentials{Token: token}) + if err != nil { + t.Fatalf("Failed to marshal AccessTokenCredentials: %v", err) + } + return a +} + +func (s) TestParse(t *testing.T) { + insecurePlugin := &anypb.Any{TypeUrl: insecureCredsTypeURL} + xdsWithInsecureFallback, err := anypb.New(&xdscredspb.XdsCredentials{FallbackCredentials: insecurePlugin}) + if err != nil { + t.Fatalf("Failed to marshal XdsCredentials: %v", err) + } + allowedInsecure := `{"dns:///my-service:443":{"channel_creds":[{"type":"insecure"}]}}` + + tests := []struct { + name string + gs *v3corepb.GrpcService + sc *bootstrap.ServerConfig + config *bootstrap.Config + // wantChannelCreds carries only the expected credentials identity; + // comparisons use Equal, which compares identities. + wantChannelCreds *creds.ChannelCreds + wantErr string + }{ + { + name: "trusted_insecure_channel_creds", + gs: googleGrpcService(target, []*anypb.Any{insecurePlugin}, nil), + sc: trustedServerConfig(t), + config: bootstrapConfig(t, "{}"), + wantChannelCreds: creds.NewChannelCreds(nil, creds.NewProtoIdentity(insecurePlugin), nil), + }, + { + name: "trusted_xds_creds_resolve_to_fallback", + gs: googleGrpcService(target, []*anypb.Any{xdsWithInsecureFallback}, nil), + sc: trustedServerConfig(t), + config: bootstrapConfig(t, "{}"), + wantChannelCreds: creds.NewChannelCreds(nil, creds.NewProtoIdentity(xdsWithInsecureFallback), nil), + }, + { + name: "trusted_no_supported_channel_creds", + gs: googleGrpcService(target, nil, nil), + sc: trustedServerConfig(t), + config: bootstrapConfig(t, "{}"), + wantErr: "no supported channel credentials", + }, + { + name: "untrusted_allowlisted_uses_allowlist_creds", + gs: googleGrpcService(target, []*anypb.Any{insecurePlugin}, nil), + config: bootstrapConfig(t, allowedInsecure), + wantChannelCreds: creds.NewChannelCreds(nil, creds.NewJSONIdentity("insecure", nil), nil), + }, + { + name: "untrusted_not_allowlisted", + gs: googleGrpcService(target, nil, nil), + config: bootstrapConfig(t, "{}"), + wantErr: "not present in allowed_grpc_services", + }, + { + name: "untrusted_nil_bootstrap_config", + gs: googleGrpcService(target, []*anypb.Any{insecurePlugin}, nil), + config: nil, + wantErr: "not present in allowed_grpc_services", + }, + { + name: "missing_google_grpc", + gs: &v3corepb.GrpcService{}, + sc: trustedServerConfig(t), + config: bootstrapConfig(t, "{}"), + wantErr: "only google_grpc", + }, + { + name: "empty_target_uri", + gs: googleGrpcService("", nil, nil), + sc: trustedServerConfig(t), + config: bootstrapConfig(t, "{}"), + wantErr: "target_uri must be non-empty", + }, + { + name: "zero_timeout_rejected", + gs: googleGrpcService(target, []*anypb.Any{insecurePlugin}, durationpb.New(0)), + sc: trustedServerConfig(t), + config: bootstrapConfig(t, "{}"), + wantErr: "timeout must be strictly positive", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got, err := Parse(test.gs, test.config, test.sc) + if test.wantErr != "" { + if err == nil || !strings.Contains(err.Error(), test.wantErr) { + t.Fatalf("Parse() error = %v, want substring %q", err, test.wantErr) + } + return + } + if err != nil { + t.Fatalf("Parse() returned unexpected error: %v", err) + } + if got.TargetURI != target { + t.Errorf("Parse() TargetURI = %q, want %q", got.TargetURI, target) + } + if got.ChannelCredentials.Bundle() == nil { + t.Error("Parse() returned channel credentials without a built bundle") + } + if !got.ChannelCredentials.Equal(test.wantChannelCreds) { + t.Errorf("Parse() ChannelCredentials identity mismatch, got %+v", got.ChannelCredentials) + } + }) + } +} + +func (s) TestParseCallCredentials(t *testing.T) { + insecurePlugin := &anypb.Any{TypeUrl: insecureCredsTypeURL} + tokenPlugin := accessTokenPlugin(t, "test-token") + sc := trustedServerConfig(t) + bc := bootstrapConfig(t, "{}") + + gs := googleGrpcService(target, []*anypb.Any{insecurePlugin}, nil) + gs.GetGoogleGrpc().CallCredentialsPlugin = []*anypb.Any{ + tokenPlugin, + // Plugins of unsupported types are skipped. + {TypeUrl: "type.googleapis.com/unsupported.CallCredentials"}, + } + got, err := Parse(gs, bc, sc) + if err != nil { + t.Fatalf("Parse() returned unexpected error: %v", err) + } + if len(got.CallCredentials) != 1 { + t.Fatalf("Parse() returned %d call credentials, want 1", len(got.CallCredentials)) + } + if got.CallCredentials[0].Credentials() == nil { + t.Error("Parse() returned call credentials without built credentials") + } + if want := creds.NewCallCreds(nil, creds.NewProtoIdentity(tokenPlugin), nil); !got.CallCredentials[0].Equal(want) { + t.Errorf("Parse() CallCredentials identity mismatch, got %+v", got.CallCredentials[0]) + } + + // An empty token must be rejected. + gs.GetGoogleGrpc().CallCredentialsPlugin = []*anypb.Any{accessTokenPlugin(t, "")} + if _, err := Parse(gs, bc, sc); err == nil || !strings.Contains(err.Error(), "access token must be non-empty") { + t.Fatalf("Parse() error = %v, want substring %q", err, "access token must be non-empty") + } +} + +func (s) TestParseInitialMetadata(t *testing.T) { + gs := googleGrpcService(target, []*anypb.Any{{TypeUrl: insecureCredsTypeURL}}, nil) + gs.InitialMetadata = []*v3corepb.HeaderValue{ + {Key: "key-b", Value: "b"}, + {Key: "key-a", Value: "legacy", RawValue: []byte("raw-a")}, + } + got, err := Parse(gs, bootstrapConfig(t, "{}"), trustedServerConfig(t)) + if err != nil { + t.Fatalf("Parse() returned unexpected error: %v", err) + } + want := metadata.MD{"key-b": []string{"b"}, "key-a": []string{"raw-a"}} + if diff := cmp.Diff(want, got.InitialMetadata); diff != "" { + t.Errorf("Parse() InitialMetadata mismatch (-want +got):\n%s", diff) + } +} + +func (s) TestConfigEqual(t *testing.T) { + insecurePlugin := &anypb.Any{TypeUrl: insecureCredsTypeURL} + protoInsecure := creds.NewChannelCreds(nil, creds.NewProtoIdentity(insecurePlugin), nil) + jsonInsecure := creds.NewChannelCreds(nil, creds.NewJSONIdentity("insecure", nil), nil) + + tests := []struct { + name string + a, b *Config + want bool + }{ + { + name: "equal_identities_share", + a: &Config{TargetURI: target, ChannelCredentials: protoInsecure}, + b: &Config{TargetURI: target, ChannelCredentials: creds.NewChannelCreds(nil, creds.NewProtoIdentity(insecurePlugin), nil)}, + want: true, + }, + { + name: "timeout_and_metadata_do_not_affect_sharing", + a: &Config{TargetURI: target, ChannelCredentials: protoInsecure, Timeout: 1}, + b: &Config{TargetURI: target, ChannelCredentials: protoInsecure, InitialMetadata: metadata.Pairs("k", "v")}, + want: true, + }, + { + name: "different_targets", + a: &Config{TargetURI: target, ChannelCredentials: protoInsecure}, + b: &Config{TargetURI: "dns:///other:443", ChannelCredentials: protoInsecure}, + want: false, + }, + { + name: "different_identity_flavors", + a: &Config{TargetURI: target, ChannelCredentials: protoInsecure}, + b: &Config{TargetURI: target, ChannelCredentials: jsonInsecure}, + want: false, + }, + { + name: "different_call_creds", + a: &Config{TargetURI: target, ChannelCredentials: protoInsecure}, + b: &Config{TargetURI: target, ChannelCredentials: protoInsecure, CallCredentials: []*creds.CallCreds{ + creds.NewCallCreds(nil, creds.NewJSONIdentity("access_token", nil), nil), + }}, + want: false, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := test.a.Equal(test.b); got != test.want { + t.Errorf("Equal() = %v, want %v", got, test.want) + } + }) + } +} diff --git a/internal/xds/httpfilter/ext_authz/config.go b/internal/xds/httpfilter/ext_authz/config.go index cacae56ae071..38b9527266cb 100644 --- a/internal/xds/httpfilter/ext_authz/config.go +++ b/internal/xds/httpfilter/ext_authz/config.go @@ -20,16 +20,16 @@ package extauthz import ( "google.golang.org/grpc/codes" + "google.golang.org/grpc/internal/xds/grpcservice" "google.golang.org/grpc/internal/xds/httpfilter" "google.golang.org/grpc/internal/xds/matcher" - "google.golang.org/grpc/internal/xds/xdsclient/xdsresource" ) // config contains the configuration for the external authorization filter. type config struct { httpfilter.FilterConfig // grpcService is the configuration for the external authorization server. - grpcService xdsresource.GRPCServiceConfig + grpcService grpcservice.Config // filterEnabled specifies the percentage of requests to be authorized by // the external authorization server. filterEnabled fraction diff --git a/internal/xds/httpfilter/ext_authz/ext_authz.go b/internal/xds/httpfilter/ext_authz/ext_authz.go index 1f28f5a369f5..ee579818f1b6 100644 --- a/internal/xds/httpfilter/ext_authz/ext_authz.go +++ b/internal/xds/httpfilter/ext_authz/ext_authz.go @@ -26,9 +26,9 @@ import ( "google.golang.org/grpc/codes" "google.golang.org/grpc/internal/envconfig" "google.golang.org/grpc/internal/transport" + "google.golang.org/grpc/internal/xds/grpcservice" "google.golang.org/grpc/internal/xds/httpfilter" "google.golang.org/grpc/internal/xds/matcher" - "google.golang.org/grpc/internal/xds/xdsclient/xdsresource" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/anypb" @@ -44,9 +44,10 @@ func init() { } var ( - // TODO: Remove this once gRFC A102 is implemented. - parseGRPCServiceConfig = func(*v3corepb.GrpcService) (xdsresource.GRPCServiceConfig, error) { - return xdsresource.GRPCServiceConfig{}, fmt.Errorf("parseGRPCServiceConfig not implemented") + // TODO: Parse via grpcservice.Parse with the filter parse options, + // as ext_proc does, when ext_authz is wired up for gRFC A102. + parseGRPCServiceConfig = func(*v3corepb.GrpcService) (grpcservice.Config, error) { + return grpcservice.Config{}, fmt.Errorf("parseGRPCServiceConfig not implemented") } ) diff --git a/internal/xds/httpfilter/ext_authz/ext_authz_test.go b/internal/xds/httpfilter/ext_authz/ext_authz_test.go index 6afbba202ba3..0a548fcbc797 100644 --- a/internal/xds/httpfilter/ext_authz/ext_authz_test.go +++ b/internal/xds/httpfilter/ext_authz/ext_authz_test.go @@ -28,9 +28,9 @@ import ( "google.golang.org/grpc/codes" "google.golang.org/grpc/internal/grpctest" "google.golang.org/grpc/internal/testutils" + "google.golang.org/grpc/internal/xds/grpcservice" "google.golang.org/grpc/internal/xds/httpfilter" "google.golang.org/grpc/internal/xds/matcher" - "google.golang.org/grpc/internal/xds/xdsclient/xdsresource" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/anypb" "google.golang.org/protobuf/types/known/wrapperspb" @@ -53,18 +53,18 @@ func Test(t *testing.T) { // testParseGRPCServiceConfig is a helper function that parses a GrpcService // proto message into a GRPCServiceConfig. This is a temporary test // implementation that will be removed once gRFC A102 is implemented. -func testParseGRPCServiceConfig(grpcService *corepb.GrpcService) (xdsresource.GRPCServiceConfig, error) { +func testParseGRPCServiceConfig(grpcService *corepb.GrpcService) (grpcservice.Config, error) { if grpcService == nil { - return xdsresource.GRPCServiceConfig{}, nil + return grpcservice.Config{}, nil } if grpcService.GetGoogleGrpc() == nil { - return xdsresource.GRPCServiceConfig{}, fmt.Errorf("only google_grpc grpc_service is supported") + return grpcservice.Config{}, fmt.Errorf("only google_grpc grpc_service is supported") } if grpcService.GetGoogleGrpc().GetTargetUri() == "" { - return xdsresource.GRPCServiceConfig{}, fmt.Errorf("targetURI must be a non-empty string") + return grpcservice.Config{}, fmt.Errorf("targetURI must be a non-empty string") } - sc := xdsresource.GRPCServiceConfig{ + sc := grpcservice.Config{ TargetURI: grpcService.GetGoogleGrpc().GetTargetUri(), } return sc, nil @@ -73,7 +73,7 @@ func testParseGRPCServiceConfig(grpcService *corepb.GrpcService) (xdsresource.GR var cmpOpts = []cmp.Option{ cmp.AllowUnexported( config{}, - xdsresource.GRPCServiceConfig{}, + grpcservice.Config{}, fraction{}, ), cmp.Transformer("RegexpToString", func(r *regexp.Regexp) string { @@ -115,7 +115,7 @@ func (s) TestParseFilterConfig_Success(t *testing.T) { }, }), wantCfg: config{ - grpcService: xdsresource.GRPCServiceConfig{ + grpcService: grpcservice.Config{ TargetURI: "localhost:1234", }, filterEnabled: fraction{ @@ -171,7 +171,7 @@ func (s) TestParseFilterConfig_Success(t *testing.T) { IncludePeerCertificate: true, }), wantCfg: config{ - grpcService: xdsresource.GRPCServiceConfig{ + grpcService: grpcservice.Config{ TargetURI: "localhost:5678", }, filterEnabled: fraction{ diff --git a/internal/xds/httpfilter/extproc/config.go b/internal/xds/httpfilter/extproc/config.go index e3300d3c1d4f..7442afe9ea15 100644 --- a/internal/xds/httpfilter/extproc/config.go +++ b/internal/xds/httpfilter/extproc/config.go @@ -22,9 +22,9 @@ import ( "time" "google.golang.org/grpc/internal/optional" + "google.golang.org/grpc/internal/xds/grpcservice" "google.golang.org/grpc/internal/xds/httpfilter" "google.golang.org/grpc/internal/xds/matcher" - "google.golang.org/grpc/internal/xds/xdsclient/xdsresource" v3procfilterpb "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/ext_proc/v3" ) @@ -38,7 +38,7 @@ type baseConfig struct { // config. If both are set, the override config will be used. // server is the configuration for the external processing server. - server xdsresource.GRPCServiceConfig + server grpcservice.Config // processingModes specifies the processing mode for each dataplane event. processingModes processingModes // failureModeAllow specifies the behavior when the RPC to the external @@ -90,7 +90,7 @@ type baseConfig struct { // base config. type overrideConfig struct { httpfilter.FilterConfig - server optional.Optional[xdsresource.GRPCServiceConfig] + server optional.Optional[grpcservice.Config] processingModes optional.Optional[processingModes] failureModeAllow optional.Optional[bool] requestAttributes []string diff --git a/internal/xds/httpfilter/extproc/config_test.go b/internal/xds/httpfilter/extproc/config_test.go index a160c8453db4..5a26cceff794 100644 --- a/internal/xds/httpfilter/extproc/config_test.go +++ b/internal/xds/httpfilter/extproc/config_test.go @@ -19,6 +19,7 @@ package extproc import ( + "encoding/json" "fmt" "regexp" "strings" @@ -28,12 +29,15 @@ import ( "github.com/google/go-cmp/cmp" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/internal/envconfig" "google.golang.org/grpc/internal/grpctest" "google.golang.org/grpc/internal/optional" + "google.golang.org/grpc/internal/testutils" + "google.golang.org/grpc/internal/xds/bootstrap" + "google.golang.org/grpc/internal/xds/grpcservice" + "google.golang.org/grpc/internal/xds/grpcservice/creds" "google.golang.org/grpc/internal/xds/httpfilter" - iextproc "google.golang.org/grpc/internal/xds/httpfilter/extproc/internal" "google.golang.org/grpc/internal/xds/matcher" - "google.golang.org/grpc/internal/xds/xdsclient/xdsresource" "google.golang.org/grpc/metadata" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/testing/protocmp" @@ -54,39 +58,81 @@ func Test(t *testing.T) { grpctest.RunSubTests(t, s{}) } -const testBaseURI = "base-uri" +const ( + testBaseURI = "base-uri" -// testParseGRPCServiceConfig is a helper function that parses a GrpcService -// proto message into a GRPCServiceConfig. This is a temporary test -// implementation that will be removed once gRFC A102 is implemented. -func testParseGRPCServiceConfig(grpcService *corepb.GrpcService) (xdsresource.GRPCServiceConfig, error) { - if grpcService == nil { - return xdsresource.GRPCServiceConfig{}, nil + insecureCredsTypeURL = "type.googleapis.com/envoy.extensions.grpc_service.channel_credentials.insecure.v3.InsecureCredentials" +) + +// allowlistInsecureCreds carries the identity of the insecure channel +// credentials configured for allowlisted targets by testParseOptions; want +// configs compare against it by identity. +var allowlistInsecureCreds = creds.NewChannelCreds(nil, creds.NewJSONIdentity("insecure", nil), nil) + +// testParseOptions returns ParseOptions whose bootstrap +// configuration allowlists the given side-channel targets with insecure +// channel credentials. The returned options carry no ServerConfig, so the +// delivering server is treated as untrusted and GrpcService parsing takes the +// allowed_grpc_services path. +func testParseOptions(t *testing.T, targets ...string) httpfilter.ParseOptions { + t.Helper() + + // The allowed_grpc_services bootstrap field is parsed only when a + // consuming feature is enabled. + testutils.SetEnvConfig(t, &envconfig.XDSClientExtProcEnabled, true) + + allowed := make(map[string]json.RawMessage, len(targets)) + for _, target := range targets { + allowed[target] = json.RawMessage(`{"channel_creds": [{"type": "insecure"}]}`) + } + allowedJSON, err := json.Marshal(allowed) + if err != nil { + t.Fatalf("Failed to marshal allowed_grpc_services: %v", err) } - if grpcService.GetGoogleGrpc() == nil { - return xdsresource.GRPCServiceConfig{}, fmt.Errorf("only google_grpc grpc_service is supported") + contents, err := bootstrap.NewContentsForTesting(bootstrap.ConfigOptionsForTesting{ + Servers: []byte(`[{"server_uri": "passthrough:///unused", "channel_creds": [{"type": "insecure"}]}]`), + Node: []byte(`{"id": "test-node"}`), + AllowedGRPCServices: allowedJSON, + }) + if err != nil { + t.Fatalf("Failed to create bootstrap contents: %v", err) } - if grpcService.GetGoogleGrpc().GetTargetUri() == "" { - return xdsresource.GRPCServiceConfig{}, fmt.Errorf("targetURI must be a non-empty string") + config, err := bootstrap.NewConfigFromContents(contents) + if err != nil { + t.Fatalf("Failed to parse bootstrap contents: %v", err) } + return httpfilter.ParseOptions{BootstrapConfig: config} +} + +// fakeSideChannelFactory implements httpfilter.SideChannelFactory. It creates +// insecure channels, and fails channel creation for failTarget. +type fakeSideChannelFactory struct { + failTarget string +} - sc := xdsresource.GRPCServiceConfig{ - TargetURI: grpcService.GetGoogleGrpc().GetTargetUri(), +func (f *fakeSideChannelFactory) CreateChannel(cfg *grpcservice.Config) (grpc.ClientConnInterface, func(), error) { + if f.failTarget != "" && cfg.TargetURI == f.failTarget { + return nil, nil, fmt.Errorf("dial error") } - return sc, nil + cc, err := grpc.NewClient(cfg.TargetURI, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + return nil, nil, err + } + return cc, func() { cc.Close() }, nil } var cmpOpts = []cmp.Option{ cmp.AllowUnexported( baseConfig{}, overrideConfig{}, - xdsresource.GRPCServiceConfig{}, processingModes{}, httpfilter.HeaderMutationRules{}, - optional.Optional[xdsresource.GRPCServiceConfig]{}, + optional.Optional[grpcservice.Config]{}, optional.Optional[processingModes]{}, optional.Optional[bool]{}, ), + cmp.Comparer((*creds.ChannelCreds).Equal), + cmp.Comparer((*creds.CallCreds).Equal), protocmp.Transform(), cmp.Transformer("RegexpToString", func(r *regexp.Regexp) string { if r == nil { @@ -100,9 +146,7 @@ var cmpOpts = []cmp.Option{ } func (s) TestParseFilterConfig_Success(t *testing.T) { - origParseGRPCServiceConfig := iextproc.ParseGRPCServiceConfig - defer func() { iextproc.ParseGRPCServiceConfig = origParseGRPCServiceConfig }() - iextproc.ParseGRPCServiceConfig = testParseGRPCServiceConfig + opts := testParseOptions(t, "localhost:1234") tests := []struct { name string @@ -125,10 +169,7 @@ func (s) TestParseFilterConfig_Success(t *testing.T) { return m }(), wantCfg: baseConfig{ - server: xdsresource.GRPCServiceConfig{ - TargetURI: "localhost:1234", - ChannelCredentials: "", - }, + server: grpcservice.Config{TargetURI: "localhost:1234", ChannelCredentials: allowlistInsecureCreds}, processingModes: processingModes{ requestHeaderMode: modeSend, responseHeaderMode: modeSend, @@ -160,10 +201,7 @@ func (s) TestParseFilterConfig_Success(t *testing.T) { return m }(), wantCfg: baseConfig{ - server: xdsresource.GRPCServiceConfig{ - TargetURI: "localhost:1234", - ChannelCredentials: "", - }, + server: grpcservice.Config{TargetURI: "localhost:1234", ChannelCredentials: allowlistInsecureCreds}, processingModes: processingModes{ requestHeaderMode: modeSend, responseHeaderMode: modeSend, @@ -195,10 +233,7 @@ func (s) TestParseFilterConfig_Success(t *testing.T) { return m }(), wantCfg: baseConfig{ - server: xdsresource.GRPCServiceConfig{ - TargetURI: "localhost:1234", - ChannelCredentials: "", - }, + server: grpcservice.Config{TargetURI: "localhost:1234", ChannelCredentials: allowlistInsecureCreds}, processingModes: processingModes{ requestHeaderMode: modeSend, responseHeaderMode: modeSend, @@ -219,7 +254,7 @@ func (s) TestParseFilterConfig_Success(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { b := builder{} - got, err := b.ParseFilterConfig(tt.cfg, httpfilter.ParseOptions{}) + got, err := b.ParseFilterConfig(tt.cfg, opts) if err != nil { t.Fatalf("ParseFilterConfig() returned unexpected error: %v", err) } @@ -230,10 +265,95 @@ func (s) TestParseFilterConfig_Success(t *testing.T) { } } +// Tests the gRFC A102 trust policy applied when parsing the grpc_service: +// credentials from the proto are honored only when the xDS management server +// that delivered the resource is trusted; for untrusted management servers +// the target must be present in the bootstrap allowed_grpc_services map, +// whose credentials are used instead. +func (s) TestParseFilterConfig_TrustPolicy(t *testing.T) { + trustedServer, err := bootstrap.ServerConfigForTesting(bootstrap.ServerConfigTestingOptions{ + URI: "trusted-server:1234", + ServerFeatures: []string{"trusted_xds_server"}, + }) + if err != nil { + t.Fatalf("ServerConfigForTesting() failed: %v", err) + } + untrustedOpts := testParseOptions(t, "localhost:1234") + trustedOpts := httpfilter.ParseOptions{BootstrapConfig: untrustedOpts.BootstrapConfig, ServerConfig: trustedServer} + + extProcConfig := func(targetURI string, channelPlugins ...*anypb.Any) proto.Message { + m, _ := anypb.New(&fpb.ExternalProcessor{ + GrpcService: &corepb.GrpcService{ + TargetSpecifier: &corepb.GrpcService_GoogleGrpc_{ + GoogleGrpc: &corepb.GrpcService_GoogleGrpc{ + TargetUri: targetURI, + ChannelCredentialsPlugin: channelPlugins, + }, + }, + }, + ProcessingMode: &fpb.ProcessingMode{}, + }) + return m + } + insecurePlugin := &anypb.Any{TypeUrl: insecureCredsTypeURL} + + tests := []struct { + name string + cfg proto.Message + opts httpfilter.ParseOptions + // wantCreds carries only the expected credentials identity; + // comparisons use Equal, which compares identities. + wantCreds *creds.ChannelCreds + wantErr string + }{ + { + name: "trusted_uses_proto_creds", + cfg: extProcConfig("localhost:1234", insecurePlugin), + opts: trustedOpts, + wantCreds: creds.NewChannelCreds(nil, creds.NewProtoIdentity(insecurePlugin), nil), + }, + { + name: "trusted_requires_supported_creds", + cfg: extProcConfig("localhost:1234"), + opts: trustedOpts, + wantErr: "no supported channel credentials found", + }, + { + name: "untrusted_allowlisted_uses_allowlist_creds", + cfg: extProcConfig("localhost:1234", insecurePlugin), + opts: untrustedOpts, + // The proto's credentials must be ignored in favor of the + // allowlisted (bootstrap JSON) ones. + wantCreds: allowlistInsecureCreds, + }, + { + name: "untrusted_not_allowlisted", + cfg: extProcConfig("other-target:1234", insecurePlugin), + opts: untrustedOpts, + wantErr: "not present in allowed_grpc_services", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := builder{}.ParseFilterConfig(tt.cfg, tt.opts) + if tt.wantErr != "" { + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("ParseFilterConfig() returned error = %v, wantErr %v", err, tt.wantErr) + } + return + } + if err != nil { + t.Fatalf("ParseFilterConfig() returned unexpected error: %v", err) + } + if gotCreds := got.(baseConfig).server.ChannelCredentials; !gotCreds.Equal(tt.wantCreds) { + t.Fatalf("ParseFilterConfig() returned channel credentials %+v, want %+v", gotCreds, tt.wantCreds) + } + }) + } +} + func (s) TestParseFilterConfig_Errors(t *testing.T) { - origParseGRPCServiceConfig := iextproc.ParseGRPCServiceConfig - defer func() { iextproc.ParseGRPCServiceConfig = origParseGRPCServiceConfig }() - iextproc.ParseGRPCServiceConfig = testParseGRPCServiceConfig + opts := testParseOptions(t, "localhost:1234") tests := []struct { name string @@ -263,7 +383,7 @@ func (s) TestParseFilterConfig_Errors(t *testing.T) { }) return m }(), - wantErr: "extproc: failed to parse grpc_service only google_grpc grpc_service is supported", + wantErr: "only google_grpc GrpcService config is supported", }, { name: "MissingProcessingMode", @@ -416,7 +536,7 @@ func (s) TestParseFilterConfig_Errors(t *testing.T) { }) return m }(), - wantErr: "extproc: failed to parse grpc_service targetURI must be a non-empty string", + wantErr: "target_uri must be non-empty", }, { name: "InvalidConfigType", @@ -427,7 +547,7 @@ func (s) TestParseFilterConfig_Errors(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { builder := builder{} - _, err := builder.ParseFilterConfig(tt.cfg, httpfilter.ParseOptions{}) + _, err := builder.ParseFilterConfig(tt.cfg, opts) if err == nil || !strings.Contains(err.Error(), tt.wantErr) { t.Fatalf("ParseFilterConfig() returned error = %v, wantErr %v", err, tt.wantErr) } @@ -600,13 +720,6 @@ func (s) TestParseFilterConfigOverride_Errors(t *testing.T) { } func (s) TestBuildClientInterceptor_Success(t *testing.T) { - origCreateExtProcChannel := iextproc.CreateExtProcChannel - iextproc.CreateExtProcChannel = func(cfg xdsresource.GRPCServiceConfig) (grpc.ClientConnInterface, func() error, error) { - conn, _ := grpc.NewClient(cfg.TargetURI, grpc.WithTransportCredentials(insecure.NewCredentials())) - return conn, conn.Close, nil - } - defer func() { iextproc.CreateExtProcChannel = origCreateExtProcChannel }() - tests := []struct { name string cfg httpfilter.FilterConfig @@ -629,10 +742,10 @@ func (s) TestBuildClientInterceptor_Success(t *testing.T) { requestBodyMode: modeSend, responseBodyMode: modeSkip, }, - server: xdsresource.GRPCServiceConfig{ + server: grpcservice.Config{ TargetURI: testBaseURI, - ChannelCredentials: "test-channel-creds", - CallCredentials: "test-call-creds", + ChannelCredentials: creds.NewChannelCreds(nil, creds.NewJSONIdentity("test-channel-creds", nil), nil), + CallCredentials: []*creds.CallCreds{creds.NewCallCreds(nil, creds.NewJSONIdentity("test-call-creds", nil), nil)}, InitialMetadata: metadata.MD(metadata.Pairs("key1", "value1")), Timeout: 5 * time.Second, }, @@ -664,10 +777,10 @@ func (s) TestBuildClientInterceptor_Success(t *testing.T) { requestBodyMode: modeSend, responseBodyMode: modeSkip, }, - server: xdsresource.GRPCServiceConfig{ + server: grpcservice.Config{ TargetURI: testBaseURI, - ChannelCredentials: "test-channel-creds", - CallCredentials: "test-call-creds", + ChannelCredentials: creds.NewChannelCreds(nil, creds.NewJSONIdentity("test-channel-creds", nil), nil), + CallCredentials: []*creds.CallCreds{creds.NewCallCreds(nil, creds.NewJSONIdentity("test-call-creds", nil), nil)}, InitialMetadata: metadata.MD(metadata.Pairs("key1", "value1")), Timeout: 5 * time.Second, }, @@ -689,7 +802,7 @@ func (s) TestBuildClientInterceptor_Success(t *testing.T) { requestBodyMode: modeSend, responseBodyMode: modeSkip, }, - server: xdsresource.GRPCServiceConfig{ + server: grpcservice.Config{ TargetURI: testBaseURI, Timeout: time.Second, InitialMetadata: metadata.MD(metadata.Pairs("key1", "value1")), @@ -715,7 +828,7 @@ func (s) TestBuildClientInterceptor_Success(t *testing.T) { requestBodyMode: modeSkip, responseBodyMode: modeSend, }), - server: optional.New(xdsresource.GRPCServiceConfig{ + server: optional.New(grpcservice.Config{ TargetURI: "override-uri", }), }, @@ -739,7 +852,7 @@ func (s) TestBuildClientInterceptor_Success(t *testing.T) { requestBodyMode: modeSkip, responseBodyMode: modeSend, }, - server: xdsresource.GRPCServiceConfig{ + server: grpcservice.Config{ TargetURI: "override-uri", }, allowedHeaders: []matcher.StringMatcher{matcher.NewExactStringMatcher("allow-header", false)}, @@ -762,7 +875,7 @@ func (s) TestBuildClientInterceptor_Success(t *testing.T) { requestBodyMode: modeSend, responseBodyMode: modeSkip, }, - server: xdsresource.GRPCServiceConfig{ + server: grpcservice.Config{ TargetURI: testBaseURI, Timeout: time.Second, InitialMetadata: metadata.MD(metadata.Pairs("key1", "value1")), @@ -799,7 +912,7 @@ func (s) TestBuildClientInterceptor_Success(t *testing.T) { requestBodyMode: modeSend, responseBodyMode: modeSkip, }, - server: xdsresource.GRPCServiceConfig{ + server: grpcservice.Config{ TargetURI: testBaseURI, Timeout: time.Second, InitialMetadata: metadata.MD(metadata.Pairs("key1", "value1")), @@ -812,7 +925,7 @@ func (s) TestBuildClientInterceptor_Success(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { builder := builder{} - filter := builder.BuildClientFilter(httpfilter.ClientFilterOptions{}) + filter := builder.BuildClientFilter(httpfilter.ClientFilterOptions{SideChannelFactory: &fakeSideChannelFactory{}}) defer filter.Close() intptr, err := filter.BuildClientInterceptor(tc.cfg, tc.override) @@ -829,16 +942,6 @@ func (s) TestBuildClientInterceptor_Success(t *testing.T) { } func (s) TestBuildClientInterceptor_Failure(t *testing.T) { - origCreateExtProcChannel := iextproc.CreateExtProcChannel - iextproc.CreateExtProcChannel = func(cfg xdsresource.GRPCServiceConfig) (grpc.ClientConnInterface, func() error, error) { - if cfg.TargetURI == "error-uri" { - return nil, nil, fmt.Errorf("dial error") - } - conn, _ := grpc.NewClient(cfg.TargetURI, grpc.WithTransportCredentials(insecure.NewCredentials())) - return conn, conn.Close, nil - } - defer func() { iextproc.CreateExtProcChannel = origCreateExtProcChannel }() - // incorrectFilterConfig embeds httpfilter.FilterConfig but is not of type // baseConfig/overrideConfig, and is used to test incorrect config types being // passed to BuildClientInterceptor. @@ -871,7 +974,7 @@ func (s) TestBuildClientInterceptor_Failure(t *testing.T) { { name: "ChannelCreationFailure", cfg: baseConfig{ - server: xdsresource.GRPCServiceConfig{ + server: grpcservice.Config{ TargetURI: "error-uri", }, }, @@ -880,12 +983,12 @@ func (s) TestBuildClientInterceptor_Failure(t *testing.T) { { name: "ChannelCreationFailureInOverride", cfg: baseConfig{ - server: xdsresource.GRPCServiceConfig{ + server: grpcservice.Config{ TargetURI: testBaseURI, }, }, override: overrideConfig{ - server: optional.New(xdsresource.GRPCServiceConfig{ + server: optional.New(grpcservice.Config{ TargetURI: "error-uri", }), }, @@ -895,7 +998,7 @@ func (s) TestBuildClientInterceptor_Failure(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { builder := builder{} - filter := builder.BuildClientFilter(httpfilter.ClientFilterOptions{}) + filter := builder.BuildClientFilter(httpfilter.ClientFilterOptions{SideChannelFactory: &fakeSideChannelFactory{failTarget: "error-uri"}}) defer filter.Close() _, err := filter.BuildClientInterceptor(tc.cfg, tc.override) diff --git a/internal/xds/httpfilter/extproc/ext_proc.go b/internal/xds/httpfilter/extproc/ext_proc.go index 2afe59b29359..716ce3527876 100644 --- a/internal/xds/httpfilter/extproc/ext_proc.go +++ b/internal/xds/httpfilter/extproc/ext_proc.go @@ -24,6 +24,7 @@ import ( "errors" "fmt" "io" + "slices" "strings" "sync" "sync/atomic" @@ -38,10 +39,10 @@ import ( "google.golang.org/grpc/internal/grpcsync" "google.golang.org/grpc/internal/optional" "google.golang.org/grpc/internal/resolver" + "google.golang.org/grpc/internal/xds/grpcservice" "google.golang.org/grpc/internal/xds/httpfilter" iextproc "google.golang.org/grpc/internal/xds/httpfilter/extproc/internal" "google.golang.org/grpc/internal/xds/matcher" - "google.golang.org/grpc/internal/xds/xdsclient/xdsresource" "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" @@ -49,6 +50,7 @@ import ( "google.golang.org/protobuf/types/known/anypb" "google.golang.org/protobuf/types/known/structpb" + v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" v3procfilterpb "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/ext_proc/v3" v3procservicegrpc "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3" v3procservicepb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3" @@ -107,7 +109,23 @@ func validateBodyProcessingMode(mode *v3procfilterpb.ProcessingMode) error { return nil } -func (builder) ParseFilterConfig(cfg proto.Message, _ httpfilter.ParseOptions) (httpfilter.FilterConfig, error) { +// parseGrpcService parses the GrpcService proto identifying the external +// processor server. The gRFC A102 trust policy — honoring the proto's +// credentials only when the delivering xDS management server is trusted, and +// requiring an untrusted server's target to be present in the bootstrap +// allowed_grpc_services map — is applied by grpcservice.Parse. +func parseGrpcService(gs *v3corepb.GrpcService, opts httpfilter.ParseOptions) (grpcservice.Config, error) { + cfg, err := grpcservice.Parse(gs, opts.BootstrapConfig, opts.ServerConfig) + if err != nil { + return grpcservice.Config{}, err + } + return *cfg, nil +} + +// ParseFilterConfig parses the provided filter configuration. The GrpcService +// identifying the external processor server is validated against the provided +// parse options, as per gRFC A102. +func (builder) ParseFilterConfig(cfg proto.Message, opts httpfilter.ParseOptions) (httpfilter.FilterConfig, error) { m, ok := cfg.(*anypb.Any) if !ok { return nil, fmt.Errorf("extproc: error parsing config %v: unknown type %T, want *anypb.Any", cfg, cfg) @@ -126,9 +144,9 @@ func (builder) ParseFilterConfig(cfg proto.Message, _ httpfilter.ParseOptions) ( if msg.GetGrpcService() == nil { return nil, fmt.Errorf("extproc: empty grpc_service provided in config %v", cfg) } - server, err := iextproc.ParseGRPCServiceConfig(msg.GetGrpcService()) + server, err := parseGrpcService(msg.GetGrpcService(), opts) if err != nil { - return nil, fmt.Errorf("extproc: failed to parse grpc_service %v", err) + return nil, fmt.Errorf("extproc: failed to parse grpc_service: %v", err) } mutationRules, err := httpfilter.HeaderMutationRulesFromProto(msg.GetMutationRules()) @@ -171,7 +189,10 @@ func (builder) ParseFilterConfig(cfg proto.Message, _ httpfilter.ParseOptions) ( }, nil } -func (builder) ParseFilterConfigOverride(ov proto.Message, _ httpfilter.ParseOptions) (httpfilter.FilterConfig, error) { +// ParseFilterConfigOverride parses the provided override filter +// configuration. The GrpcService identifying the external processor server is +// validated against the provided parse options, as per gRFC A102. +func (builder) ParseFilterConfigOverride(ov proto.Message, opts httpfilter.ParseOptions) (httpfilter.FilterConfig, error) { m, ok := ov.(*anypb.Any) if !ok { return nil, fmt.Errorf("extproc: error parsing override %v: unknown type %T, want *anypb.Any", ov, ov) @@ -190,9 +211,9 @@ func (builder) ParseFilterConfigOverride(ov proto.Message, _ httpfilter.ParseOpt processingModesOpt = optional.New(processingModesFromProto(pm)) } - var serverOpt optional.Optional[xdsresource.GRPCServiceConfig] + var serverOpt optional.Optional[grpcservice.Config] if override.GetGrpcService() != nil { - server, err := iextproc.ParseGRPCServiceConfig(override.GetGrpcService()) + server, err := parseGrpcService(override.GetGrpcService(), opts) if err != nil { return nil, fmt.Errorf("extproc: failed to parse grpc_service: %v", err) } @@ -221,114 +242,104 @@ func (builder) BuildClientFilter(opts httpfilter.ClientFilterOptions) httpfilter return &clientFilter{ metricsRecorder: opts.MetricsRecorder, target: opts.Target, - procChannels: make(map[grpcServiceKey]*grpcsync.RefCounted[v3procservicegrpc.ExternalProcessorClient]), + factory: opts.SideChannelFactory, } } var _ httpfilter.ClientFilterBuilder = builder{} -// grpcServiceKey uniquely identifies an external processor server configuration -// by its target URI, channel credentials, and call credentials. It is used as a -// map key in clientFilter to share and reuse the external processor channels. -type grpcServiceKey struct { - targetURI string - channelCredentials string - callCredentials string +// procChannelEntry is a shared external processor channel, together with the +// server config it was created from. The config is used only for equality +// comparisons when deciding whether the channel can be reused. +type procChannelEntry struct { + server grpcservice.Config + rc *grpcsync.RefCounted[v3procservicegrpc.ExternalProcessorClient] } type clientFilter struct { metricsRecorder estats.MetricsRecorder target string + factory httpfilter.SideChannelFactory mu sync.Mutex - procChannels map[grpcServiceKey]*grpcsync.RefCounted[v3procservicegrpc.ExternalProcessorClient] + procChannels []*procChannelEntry } func (*clientFilter) Close() {} -// getProcChannel returns an existing refcounted client from the map if present -// and its refcount is incremented successfully. -func (cf *clientFilter) getProcChannel(key grpcServiceKey) *grpcsync.RefCounted[v3procservicegrpc.ExternalProcessorClient] { +// getProcChannel returns an existing refcounted client for an equal server +// config if present and its refcount is incremented successfully. +func (cf *clientFilter) getProcChannel(server *grpcservice.Config) *grpcsync.RefCounted[v3procservicegrpc.ExternalProcessorClient] { cf.mu.Lock() defer cf.mu.Unlock() - if rc, ok := cf.procChannels[key]; ok && rc.TryIncrement() { - return rc + for _, e := range cf.procChannels { + if e.server.Equal(server) && e.rc.TryIncrement() { + return e.rc + } } return nil } -// storeProcChannel stores the created channel in the map if no valid channel -// exists for the key. If another goroutine already stored a channel while -// unlocked, it increments the existing channel's refcount and returns it. -func (cf *clientFilter) storeProcChannel(key grpcServiceKey, rc *grpcsync.RefCounted[v3procservicegrpc.ExternalProcessorClient]) *grpcsync.RefCounted[v3procservicegrpc.ExternalProcessorClient] { +// storeProcChannel stores the created channel entry if no valid channel +// exists for an equal server config. If another goroutine already stored one +// while unlocked, it increments the existing channel's refcount and returns +// it. +func (cf *clientFilter) storeProcChannel(entry *procChannelEntry) *grpcsync.RefCounted[v3procservicegrpc.ExternalProcessorClient] { cf.mu.Lock() defer cf.mu.Unlock() - if existing, ok := cf.procChannels[key]; ok && existing.TryIncrement() { - return existing + for _, e := range cf.procChannels { + if e.server.Equal(&entry.server) && e.rc.TryIncrement() { + return e.rc + } } - cf.procChannels[key] = rc - return rc + cf.procChannels = append(cf.procChannels, entry) + return entry.rc } -// removeProcChannel removes rc from the map if it is still associated with key. -// -// We check (cf.procChannels[key] == rc) before deleting because: -// 1. When a channel's reference count drops to 0, this cleanup callback runs -// asynchronously on a background goroutine. -// 2. Before this callback acquires cf.mu, a subsequent call to -// getOrCreateExtProcChannel could see that the old channel has refcount 0, -// ignore it, and store a newly created channel in cf.procChannels[key]. -// 3. The equality check ensures we only delete from the map if it still points -// to this expiring channel, preventing us from accidentally deleting a new -// replacement channel. -func (cf *clientFilter) removeProcChannel(key grpcServiceKey, rc *grpcsync.RefCounted[v3procservicegrpc.ExternalProcessorClient]) { +// removeProcChannel removes the entry from the list. When a channel's +// reference count drops to zero this cleanup runs asynchronously, so a +// replacement entry with an equal config may already have been stored; +// removal by entry pointer ensures only the expiring entry is removed. +func (cf *clientFilter) removeProcChannel(entry *procChannelEntry) { cf.mu.Lock() defer cf.mu.Unlock() - // Only delete from the map if it hasn't already been replaced by a newer channel. - if cf.procChannels[key] == rc { - delete(cf.procChannels, key) - } + cf.procChannels = slices.DeleteFunc(cf.procChannels, func(e *procChannelEntry) bool { return e == entry }) } // getOrCreateExtProcChannel retrieves an existing refcounted external processor -// client from the procChannels map and increases its refcount or creates a new -// one if it doesn't exist. -func (cf *clientFilter) getOrCreateExtProcChannel(server xdsresource.GRPCServiceConfig) (*grpcsync.RefCounted[v3procservicegrpc.ExternalProcessorClient], error) { - // Create the grpcServiceKey. - key := grpcServiceKey{ - targetURI: server.TargetURI, - channelCredentials: server.ChannelCredentials, - callCredentials: server.CallCredentials, - } - - // If the channel for the key is present in the map and its refcount is +// client for an equal server config and increases its refcount, or creates a +// new one if there is none. +func (cf *clientFilter) getOrCreateExtProcChannel(server grpcservice.Config) (*grpcsync.RefCounted[v3procservicegrpc.ExternalProcessorClient], error) { + // If a channel for an equal server config is present and its refcount is // greater than 0, increment the refcount and return the channel. - if rc := cf.getProcChannel(key); rc != nil { + if rc := cf.getProcChannel(&server); rc != nil { return rc, nil } - // Create the external processor channel without holding the lock. - cc, cancel, err := iextproc.CreateExtProcChannel(server) + // Create the external processor channel without holding the lock. The + // channel is shared with other consumers via the xDS client; the release + // function drops this filter's reference to it. + cc, release, err := iextproc.CreateExtProcChannel(cf.factory, &server) if err != nil { return nil, fmt.Errorf("extproc: failed to create channel to the external processor server %q: %v", server.TargetURI, err) } client := v3procservicegrpc.NewExternalProcessorClient(cc) - // Create a new refcounted client. The onZero cleanup function will remove the - // client from the map and close the underlying channel. - var rc *grpcsync.RefCounted[v3procservicegrpc.ExternalProcessorClient] - rc = grpcsync.NewRefCounted(client, func() { - cf.removeProcChannel(key, rc) - cancel() + // Create a new refcounted client. The onZero cleanup function will remove + // the entry from the list and release the underlying channel. + entry := &procChannelEntry{server: server} + entry.rc = grpcsync.NewRefCounted(client, func() { + cf.removeProcChannel(entry) + release() }) - // Double-check if another goroutine created and stored a channel for this - // key while we were unlocked. - if existing := cf.storeProcChannel(key, rc); existing != rc { - rc.Decrement() + // Double-check if another goroutine created and stored a channel for an + // equal config while we were unlocked. + if existing := cf.storeProcChannel(entry); existing != entry.rc { + entry.rc.Decrement() return existing, nil } - return rc, nil + return entry.rc, nil } func (cf *clientFilter) BuildClientInterceptor(base, override httpfilter.FilterConfig) (httpfilter.ClientInterceptor, error) { @@ -347,6 +358,9 @@ func (cf *clientFilter) BuildClientInterceptor(base, override httpfilter.FilterC config := newInterceptorConfig(b, ov) + if cf.factory == nil { + return nil, fmt.Errorf("extproc: no side-channel factory provided to create a channel to the external processor server") + } // Create or reuse a refcounted channel to the external processor server. rc, err := cf.getOrCreateExtProcChannel(config.server) if err != nil { @@ -375,7 +389,7 @@ func (i *clientInterceptor) Close() { // processor server to be able to cancel it independently. This context has a // deadline of the timeout specified in the config, if present, and contains the // initial metadata specified in the config. -func createProcContext(ctx context.Context, server xdsresource.GRPCServiceConfig) (procCtx context.Context, cancel context.CancelFunc) { +func createProcContext(ctx context.Context, server grpcservice.Config) (procCtx context.Context, cancel context.CancelFunc) { if server.Timeout != 0 { procCtx, cancel = context.WithTimeout(ctx, server.Timeout) } else { diff --git a/internal/xds/httpfilter/extproc/ext_proc_ext_test.go b/internal/xds/httpfilter/extproc/ext_proc_ext_test.go index 17091f798992..f3db9bb8cf3a 100644 --- a/internal/xds/httpfilter/extproc/ext_proc_ext_test.go +++ b/internal/xds/httpfilter/extproc/ext_proc_ext_test.go @@ -28,9 +28,11 @@ import ( "testing" "time" + "github.com/google/uuid" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" + grpcinternal "google.golang.org/grpc/internal" "google.golang.org/grpc/internal/envconfig" "google.golang.org/grpc/internal/grpcsync" "google.golang.org/grpc/internal/grpctest" @@ -39,11 +41,15 @@ import ( "google.golang.org/grpc/internal/testutils/stats" "google.golang.org/grpc/internal/testutils/xds/e2e" "google.golang.org/grpc/internal/testutils/xds/e2e/setup" + "google.golang.org/grpc/internal/xds/bootstrap" + "google.golang.org/grpc/internal/xds/grpcservice" + "google.golang.org/grpc/internal/xds/httpfilter" "google.golang.org/grpc/internal/xds/httpfilter/extproc/internal" - "google.golang.org/grpc/internal/xds/xdsclient/xdsresource" "google.golang.org/grpc/metadata" + "google.golang.org/grpc/resolver" "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/anypb" "google.golang.org/protobuf/types/known/durationpb" v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" @@ -73,33 +79,11 @@ const ( reqBodyC2 = "c2" respBodyS1 = "s1" respBodyS2 = "s2" -) -func parseGRPCServiceConfigForTesting(gs *v3corepb.GrpcService) (xdsresource.GRPCServiceConfig, error) { - if gs == nil { - return xdsresource.GRPCServiceConfig{}, fmt.Errorf("expected non-nil GrpcService") - } - gg := gs.GetGoogleGrpc() - if gg == nil { - return xdsresource.GRPCServiceConfig{}, fmt.Errorf("expected non-nil GoogleGrpc") - } - target := gg.GetTargetUri() - if target == "" { - return xdsresource.GRPCServiceConfig{}, fmt.Errorf("empty target_uri in GoogleGrpc") - } - return xdsresource.GRPCServiceConfig{ - TargetURI: target, - Timeout: gs.GetTimeout().AsDuration(), - }, nil -} - -func createExtProcChannelForTesting(cfg xdsresource.GRPCServiceConfig) (grpc.ClientConnInterface, func() error, error) { - cc, err := grpc.NewClient(cfg.TargetURI, grpc.WithTransportCredentials(insecure.NewCredentials())) - if err != nil { - return nil, nil, err - } - return cc, cc.Close, nil -} + // insecureCredsTypeURL identifies the insecure channel credentials + // plugin in a GrpcService proto (gRFC A102). + insecureCredsTypeURL = "type.googleapis.com/envoy.extensions.grpc_service.channel_credentials.insecure.v3.InsecureCredentials" +) func requestHeadersResponse(setHeaders map[string]string, removeHeaders []string) *v3procservicepb.ProcessingResponse { var setOptions []*v3corepb.HeaderValueOption @@ -249,19 +233,9 @@ func (s *testExtProcServer) Process(stream v3procservicegrpc.ExternalProcessor_P func startTestExtProcessor(t *testing.T, processFunc func(v3procservicegrpc.ExternalProcessor_ProcessServer) error) (string, func()) { t.Helper() - origParse := internal.ParseGRPCServiceConfig - origCreate := internal.CreateExtProcChannel - internal.ParseGRPCServiceConfig = parseGRPCServiceConfigForTesting - internal.CreateExtProcChannel = createExtProcChannelForTesting - testutils.SetEnvConfig(t, &envconfig.XDSClientExtProcEnabled, true) internal.RegisterForTesting() - - t.Cleanup(func() { - internal.ParseGRPCServiceConfig = origParse - internal.CreateExtProcChannel = origCreate - internal.UnregisterForTesting() - }) + t.Cleanup(internal.UnregisterForTesting) lis, err := testutils.LocalTCPListener() if err != nil { @@ -282,6 +256,14 @@ func startTestExtProcessor(t *testing.T, processFunc func(v3procservicegrpc.Exte func setupTestClient(t *testing.T, extProcAddr string, extProcConfig *v3procfilterpb.ExternalProcessor, serverAddr string, opts ...grpc.DialOption) (*grpc.ClientConn, error) { t.Helper() managementServer, nodeID, _, xdsResolver := setup.ManagementServerAndResolver(t) + return setupTestClientWithResolver(t, managementServer, nodeID, xdsResolver, extProcAddr, extProcConfig, serverAddr, opts...) +} + +// setupTestClientWithResolver is like setupTestClient, but uses the provided +// management server and xDS resolver instead of the default trusted-server +// setup. +func setupTestClientWithResolver(t *testing.T, managementServer *e2e.ManagementServer, nodeID string, xdsResolver resolver.Builder, extProcAddr string, extProcConfig *v3procfilterpb.ExternalProcessor, serverAddr string, opts ...grpc.DialOption) (*grpc.ClientConn, error) { + t.Helper() const serviceName = "test-service" @@ -302,10 +284,16 @@ func setupTestClient(t *testing.T, extProcAddr string, extProcConfig *v3procfilt if extProcConfig.GrpcService != nil { timeout = extProcConfig.GrpcService.GetTimeout() } + // The GrpcService proto carries insecure channel credentials: they are + // used when the delivering management server is trusted (the default + // e2e bootstrap carries the trusted_xds_server feature), and are ignored + // in favor of the bootstrap allowed_grpc_services credentials when it is + // not. extProcConfig.GrpcService = &v3corepb.GrpcService{ TargetSpecifier: &v3corepb.GrpcService_GoogleGrpc_{ GoogleGrpc: &v3corepb.GrpcService_GoogleGrpc{ - TargetUri: extProcAddr, + TargetUri: extProcAddr, + ChannelCredentialsPlugin: []*anypb.Any{{TypeUrl: insecureCredsTypeURL}}, }, }, Timeout: timeout, @@ -4827,14 +4815,14 @@ func (s) TestExtProcChannelRetention(t *testing.T) { // Override CreateExtProcChannel to signal when the channel is closed. origCreate := internal.CreateExtProcChannel - internal.CreateExtProcChannel = func(cfg xdsresource.GRPCServiceConfig) (grpc.ClientConnInterface, func() error, error) { + internal.CreateExtProcChannel = func(_ httpfilter.SideChannelFactory, cfg *grpcservice.Config) (grpc.ClientConnInterface, func(), error) { cc, err := grpc.NewClient(cfg.TargetURI, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { return nil, nil, err } - closeFunc := func() error { + closeFunc := func() { close(closeChan) - return cc.Close() + cc.Close() } return cc, closeFunc, nil } @@ -4945,14 +4933,14 @@ func (s) TestExtProcChannelRetention_XDSConfigUpdate(t *testing.T) { // Override CreateExtProcChannel to signal when a channel is closed and // dial to correct proc server. origCreate := internal.CreateExtProcChannel - internal.CreateExtProcChannel = func(cfg xdsresource.GRPCServiceConfig) (grpc.ClientConnInterface, func() error, error) { + internal.CreateExtProcChannel = func(_ httpfilter.SideChannelFactory, cfg *grpcservice.Config) (grpc.ClientConnInterface, func(), error) { cc, err := grpc.NewClient(cfg.TargetURI, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { return nil, nil, err } - closeFunc := func() error { + closeFunc := func() { closeChan <- cfg.TargetURI - return cc.Close() + cc.Close() } return cc, closeFunc, nil } @@ -5005,7 +4993,8 @@ func (s) TestExtProcChannelRetention_XDSConfigUpdate(t *testing.T) { GrpcService: &v3corepb.GrpcService{ TargetSpecifier: &v3corepb.GrpcService_GoogleGrpc_{ GoogleGrpc: &v3corepb.GrpcService_GoogleGrpc{ - TargetUri: extProcAddr, + TargetUri: extProcAddr, + ChannelCredentialsPlugin: []*anypb.Any{{TypeUrl: insecureCredsTypeURL}}, }, }, }, @@ -5125,14 +5114,14 @@ func (s) TestExtProcChannelRetention_UnaryRPC(t *testing.T) { // Override CreateExtProcChannel to signal when the channel is closed. origCreate := internal.CreateExtProcChannel - internal.CreateExtProcChannel = func(cfg xdsresource.GRPCServiceConfig) (grpc.ClientConnInterface, func() error, error) { + internal.CreateExtProcChannel = func(_ httpfilter.SideChannelFactory, cfg *grpcservice.Config) (grpc.ClientConnInterface, func(), error) { cc, err := grpc.NewClient(cfg.TargetURI, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { return nil, nil, err } - closeFunc := func() error { + closeFunc := func() { close(closeChan) - return cc.Close() + cc.Close() } return cc, closeFunc, nil } @@ -5181,3 +5170,83 @@ func (s) TestExtProcChannelRetention_UnaryRPC(t *testing.T) { }) } } + +// TestUntrustedServerAllowedGRPCServices tests the ext_proc filter when the +// xDS server is not configured with the trusted_xds_server feature. On this +// path the credentials in the GrpcService proto are ignored, and the filter +// config is accepted only because the external processor's target is present +// in the bootstrap allowed_grpc_services map, whose credentials are used for +// the side channel (gRFC A102). Verifies that a data-plane RPC flows through +// the external processor end-to-end. +func (s) TestUntrustedServerAllowedGRPCServices(t *testing.T) { + const mutatedHeader = "request-mutated" + lisAddr, _ := startTestExtProcessor(t, func(stream v3procservicegrpc.ExternalProcessor_ProcessServer) error { + for { + req, err := stream.Recv() + if err == io.EOF { + return nil + } + if err != nil { + return err + } + if req.GetRequestHeaders() == nil { + return fmt.Errorf("unexpected message from client: %v", req) + } + if err := stream.Send(requestHeadersResponse(map[string]string{mutatedHeader: "true"}, nil)); err != nil { + return err + } + } + }) + + // Start a test backend that verifies the header mutation performed by the + // external processor. + stub := stubserver.StartTestService(t, &stubserver.StubServer{ + UnaryCallF: func(ctx context.Context, _ *testpb.SimpleRequest) (*testpb.SimpleResponse, error) { + md, ok := metadata.FromIncomingContext(ctx) + if !ok { + return nil, fmt.Errorf("server did not receive incoming metadata") + } + if vals := md.Get(mutatedHeader); len(vals) == 0 || vals[0] != "true" { + return nil, fmt.Errorf("missing or invalid %q header: %v", mutatedHeader, vals) + } + return &testpb.SimpleResponse{}, nil + }, + }) + defer stub.Stop() + + // Create bootstrap configuration without the trusted_xds_server server + // feature, and with the external processor's target present in the + // allowed_grpc_services map. + managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true}) + nodeID := uuid.New().String() + bc, err := bootstrap.NewContentsForTesting(bootstrap.ConfigOptionsForTesting{ + Servers: []byte(fmt.Sprintf(`[{"server_uri": "passthrough:///%s", "channel_creds": [{"type": "insecure"}]}]`, managementServer.Address)), + Node: []byte(fmt.Sprintf(`{"id": %q}`, nodeID)), + AllowedGRPCServices: []byte(fmt.Sprintf(`{%q: {"channel_creds": [{"type": "insecure"}]}}`, lisAddr)), + }) + if err != nil { + t.Fatalf("Failed to create bootstrap contents: %v", err) + } + xdsResolver, err := grpcinternal.NewXDSResolverWithConfigForTesting.(func([]byte) (resolver.Builder, error))(bc) + if err != nil { + t.Fatalf("Failed to create xDS resolver: %v", err) + } + + cc, err := setupTestClientWithResolver(t, managementServer, nodeID, xdsResolver, lisAddr, &v3procfilterpb.ExternalProcessor{ + ProcessingMode: &v3procfilterpb.ProcessingMode{ + RequestHeaderMode: v3procfilterpb.ProcessingMode_SEND, + ResponseHeaderMode: v3procfilterpb.ProcessingMode_SKIP, + }, + }, stub.Address) + if err != nil { + t.Fatalf("Failed to dial: %v", err) + } + defer cc.Close() + + ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) + defer cancel() + client := testgrpc.NewTestServiceClient(cc) + if _, err := client.UnaryCall(ctx, &testpb.SimpleRequest{}); err != nil { + t.Fatalf("UnaryCall() failed: %v", err) + } +} diff --git a/internal/xds/httpfilter/extproc/internal/internal.go b/internal/xds/httpfilter/extproc/internal/internal.go index a6812f01d31b..a7ce51158eff 100644 --- a/internal/xds/httpfilter/extproc/internal/internal.go +++ b/internal/xds/httpfilter/extproc/internal/internal.go @@ -20,15 +20,21 @@ package internal import ( - "fmt" "time" - v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" "google.golang.org/grpc" - "google.golang.org/grpc/internal/xds/xdsclient/xdsresource" + "google.golang.org/grpc/internal/xds/grpcservice" + "google.golang.org/grpc/internal/xds/httpfilter" ) var ( + // CreateExtProcChannel creates the channel to the external processing + // server via the provided side-channel factory. It is a variable so that + // tests can intercept channel creation and observe its release. + CreateExtProcChannel = func(factory httpfilter.SideChannelFactory, server *grpcservice.Config) (grpc.ClientConnInterface, func(), error) { + return factory.CreateChannel(server) + } + // RegisterForTesting registers the external processor HTTP Filter for testing // purposes. RegisterForTesting func() @@ -37,18 +43,6 @@ var ( // testing purposes. UnregisterForTesting func() - // ParseGRPCServiceConfig parses the gRPC service configuration from the given - // protobuf message. - ParseGRPCServiceConfig = func(*v3corepb.GrpcService) (xdsresource.GRPCServiceConfig, error) { - return xdsresource.GRPCServiceConfig{}, fmt.Errorf("extproc: ParseGRPCServiceConfig not implemented") - } - - // CreateExtProcChannel creates a gRPC client channel to the external - // processing server. - CreateExtProcChannel = func(xdsresource.GRPCServiceConfig) (grpc.ClientConnInterface, func() error, error) { - return nil, nil, fmt.Errorf("extproc: dialing external processor server not implemented") - } - // TimeNowFunc returns the current time.Time, and can be overridden for // testing purposes. TimeNowFunc func() time.Time diff --git a/internal/xds/httpfilter/httpfilter.go b/internal/xds/httpfilter/httpfilter.go index 52cbc618ad3d..67f0931a3933 100644 --- a/internal/xds/httpfilter/httpfilter.go +++ b/internal/xds/httpfilter/httpfilter.go @@ -27,6 +27,7 @@ import ( estats "google.golang.org/grpc/experimental/stats" iresolver "google.golang.org/grpc/internal/resolver" "google.golang.org/grpc/internal/xds/bootstrap" + "google.golang.org/grpc/internal/xds/grpcservice" "google.golang.org/protobuf/proto" ) @@ -100,11 +101,25 @@ type ClientInterceptor interface { Close() } +// SideChannelFactory creates shared channels to xDS-configured side-channel +// services (gRFC A102). It is implemented by the xDS client. +type SideChannelFactory interface { + // CreateChannel returns a shared gRPC channel to the side-channel service + // described by the given config, creating it on first use. A channel is + // shared between configs with the same target and credential identities. + // The returned release function must be called when the caller is done + // with the channel; the channel is closed when the last user releases it. + // Credentials owned by the config are released when the channel is + // closed, or immediately if an equivalent channel already exists. + CreateChannel(cfg *grpcservice.Config) (grpc.ClientConnInterface, func(), error) +} + // ClientFilterOptions contains options for building a client filter. type ClientFilterOptions struct { - FilterName string // FilterName is the filter name from the xDS configuration. - MetricsRecorder estats.MetricsRecorder // MetricsRecorder is the metrics recorder to capture metrics for the filter. - Target string // Target is the target string of the channel. + FilterName string // FilterName is the filter name from the xDS configuration. + MetricsRecorder estats.MetricsRecorder // MetricsRecorder is the metrics recorder to capture metrics for the filter. + Target string // Target is the target string of the channel. + SideChannelFactory SideChannelFactory // SideChannelFactory creates channels to side-channel services. } // ClientFilterBuilder is an optional interface that a Builder can implement to diff --git a/internal/xds/resolver/xds_resolver.go b/internal/xds/resolver/xds_resolver.go index b9feb55bc8a3..2add398a7bdc 100644 --- a/internal/xds/resolver/xds_resolver.go +++ b/internal/xds/resolver/xds_resolver.go @@ -289,15 +289,19 @@ func (r *xdsResolver) Close() { if r.dm != nil { r.dm.Close() } - if r.xdsClientClose != nil { - r.xdsClientClose() - } if r.curConfigSelector != nil { r.curConfigSelector.stop() } for _, cf := range r.httpFilters { cf.Close() } + // Release the xDS client only after the filters are torn down: filters + // may hold side channels whose credentials are owned by the client's + // bootstrap config and are released when the last reference to the + // client is dropped. + if r.xdsClientClose != nil { + r.xdsClientClose() + } r.logger.Infof("Shutdown") } @@ -683,9 +687,10 @@ func (r *xdsResolver) getOrCreateClientFilter(builder httpfilter.ClientFilterBui } cf := builder.BuildClientFilter(httpfilter.ClientFilterOptions{ - FilterName: key.name, - MetricsRecorder: r.metricsRecorder, - Target: r.target, + FilterName: key.name, + MetricsRecorder: r.metricsRecorder, + Target: r.target, + SideChannelFactory: r.xdsClient, }) r.httpFilters[key] = cf return cf diff --git a/internal/xds/xdsclient/channel.go b/internal/xds/xdsclient/channel.go new file mode 100644 index 000000000000..5f818126cea1 --- /dev/null +++ b/internal/xds/xdsclient/channel.go @@ -0,0 +1,102 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package xdsclient + +import ( + "fmt" + "slices" + "sync" + + "google.golang.org/grpc" + "google.golang.org/grpc/internal/grpcsync" + "google.golang.org/grpc/internal/xds/grpcservice" +) + +// sideChannelEntry is a shared side channel in the pool, together with the +// config it was created from. The config is used only for equality +// comparisons when deciding whether a channel can be shared. +type sideChannelEntry struct { + cfg *grpcservice.Config + rc *grpcsync.RefCounted[*grpc.ClientConn] +} + +// CreateChannel returns a shared gRPC channel to the side-channel service +// described by the given config, creating it on first use. A channel is +// shared between configs that compare Equal, i.e. same target and same +// credential identities. The returned release function must be called when +// the caller is done with the channel; the channel is closed when the last +// user releases it. +// +// Credentials owned by the config are released when the channel is closed; +// the caller must not use them afterwards. If an Equal channel already +// exists and the config's credentials are a different build than the ones the +// channel was created with, the duplicates are released immediately. +func (c *clientImpl) CreateChannel(cfg *grpcservice.Config) (grpc.ClientConnInterface, func(), error) { + if cfg == nil || cfg.ChannelCredentials == nil { + return nil, nil, fmt.Errorf("xds: no channel credentials in side channel config %v", cfg) + } + + c.sideChannelsMu.Lock() + defer c.sideChannelsMu.Unlock() + for _, e := range c.sideChannels { + if e.cfg.Equal(cfg) && e.rc.TryIncrement() { + // Share the existing channel. If the caller's credentials are a + // different build than the ones the channel holds, release the + // duplicates; credential cleanups are idempotent, so this is a + // no-op when the caller passed the very same credentials again. + if e.cfg.ChannelCredentials != cfg.ChannelCredentials { + cfg.Close() + } + return e.rc.Value(), sideChannelRelease(e.rc), nil + } + // If TryIncrement failed, the entry's refcount already dropped to + // zero and it is being cleaned up: it is removed by its own cleanup, + // and a fresh channel is created below. + } + + dialOpts := []grpc.DialOption{grpc.WithCredentialsBundle(cfg.ChannelCredentials.Bundle())} + for _, cc := range cfg.CallCredentials { + dialOpts = append(dialOpts, grpc.WithPerRPCCredentials(cc.Credentials())) + } + conn, err := grpc.NewClient(cfg.TargetURI, dialOpts...) + if err != nil { + cfg.Close() + return nil, nil, fmt.Errorf("xds: failed to create side channel to %q: %v", cfg.TargetURI, err) + } + // Ownership of the config's credentials transfers to the entry: they are + // released when the channel is closed. Credentials borrowed from the + // bootstrap config carry no cleanup and are unaffected. + entry := &sideChannelEntry{cfg: cfg} + entry.rc = grpcsync.NewRefCounted(conn, func() { + c.sideChannelsMu.Lock() + c.sideChannels = slices.DeleteFunc(c.sideChannels, func(e *sideChannelEntry) bool { return e == entry }) + c.sideChannelsMu.Unlock() + conn.Close() + cfg.Close() + }) + c.sideChannels = append(c.sideChannels, entry) + return conn, sideChannelRelease(entry.rc), nil +} + +// sideChannelRelease returns an idempotent release function for the given +// channel entry. It must be called without holding sideChannelsMu, since the +// last release runs the cleanup synchronously, which acquires the mutex. +func sideChannelRelease(rc *grpcsync.RefCounted[*grpc.ClientConn]) func() { + return sync.OnceFunc(rc.Decrement) +} diff --git a/internal/xds/xdsclient/channel_test.go b/internal/xds/xdsclient/channel_test.go new file mode 100644 index 000000000000..d398f0654c43 --- /dev/null +++ b/internal/xds/xdsclient/channel_test.go @@ -0,0 +1,185 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package xdsclient + +import ( + "encoding/json" + "strings" + "testing" + + "google.golang.org/grpc" + "google.golang.org/grpc/connectivity" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/internal/xds/grpcservice" + "google.golang.org/grpc/internal/xds/grpcservice/accesstokencreds" + "google.golang.org/grpc/internal/xds/grpcservice/creds" +) + +// testChannelCreds returns paired insecure channel credentials whose identity +// is the given JSON credentials type name. cleanup may be nil. +func testChannelCreds(typ string, cleanup func()) *creds.ChannelCreds { + return creds.NewChannelCreds(insecure.NewBundle(), creds.NewJSONIdentity(typ, nil), cleanup) +} + +// Tests that CreateChannel returns the same shared channel for configs with +// equal credential identities, releases duplicate credential builds, and +// closes the channel only when all users have released it. +func (s) TestCreateChannel_Sharing(t *testing.T) { + c := &clientImpl{} + + cfg1 := &grpcservice.Config{TargetURI: "passthrough:///target", ChannelCredentials: testChannelCreds("insecure", nil)} + cc1, release1, err := c.CreateChannel(cfg1) + if err != nil { + t.Fatalf("CreateChannel() failed: %v", err) + } + + // An Equal config with a different credentials build must share the + // channel, and the duplicate build must be released immediately. + duplicateReleased := false + cfg2 := &grpcservice.Config{TargetURI: "passthrough:///target", ChannelCredentials: testChannelCreds("insecure", func() { duplicateReleased = true })} + cc2, release2, err := c.CreateChannel(cfg2) + if err != nil { + t.Fatalf("CreateChannel() failed: %v", err) + } + if cc1 != cc2 { + t.Fatalf("CreateChannel() returned different channels for equal configs") + } + if !duplicateReleased { + t.Fatalf("CreateChannel() did not release the duplicate credentials build on a shared channel") + } + + conn := cc1.(*grpc.ClientConn) + release1() + release1() // Calling release multiple times must be a no-op. + if got := conn.GetState(); got == connectivity.Shutdown { + t.Fatalf("Channel closed after releasing one of two references") + } + release2() + if got := conn.GetState(); got != connectivity.Shutdown { + t.Fatalf("Channel state after releasing all references: %v, want %v", got, connectivity.Shutdown) + } + + // A new call with an equal config must create a fresh channel. + cfg3 := &grpcservice.Config{TargetURI: "passthrough:///target", ChannelCredentials: testChannelCreds("insecure", nil)} + cc3, release3, err := c.CreateChannel(cfg3) + if err != nil { + t.Fatalf("CreateChannel() failed: %v", err) + } + defer release3() + if cc3 == cc1 { + t.Fatalf("CreateChannel() returned a released channel") + } +} + +// Tests that credentials owned by the config are released when the channel is +// closed. +func (s) TestCreateChannel_OwnershipTransfer(t *testing.T) { + c := &clientImpl{} + + released := false + cfg := &grpcservice.Config{TargetURI: "passthrough:///target", ChannelCredentials: testChannelCreds("insecure", func() { released = true })} + _, release, err := c.CreateChannel(cfg) + if err != nil { + t.Fatalf("CreateChannel() failed: %v", err) + } + if released { + t.Fatalf("CreateChannel() released the config's credentials while the channel is in use") + } + release() + if !released { + t.Fatalf("CreateChannel() did not release the config's credentials when the channel was closed") + } +} + +// Tests that CreateChannel returns different channels for the same target +// when the credential identities differ. +func (s) TestCreateChannel_DifferentCreds(t *testing.T) { + c := &clientImpl{} + + cc1, release1, err := c.CreateChannel(&grpcservice.Config{TargetURI: "passthrough:///target", ChannelCredentials: testChannelCreds("insecure", nil)}) + if err != nil { + t.Fatalf("CreateChannel() failed: %v", err) + } + defer release1() + cc2, release2, err := c.CreateChannel(&grpcservice.Config{TargetURI: "passthrough:///target", ChannelCredentials: testChannelCreds("other", nil)}) + if err != nil { + t.Fatalf("CreateChannel() failed: %v", err) + } + defer release2() + if cc1 == cc2 { + t.Fatalf("CreateChannel() returned the same channel for different credential identities") + } +} + +// Tests that CreateChannel fails on configs without channel credentials, and +// that a failure to create the channel releases the config's credentials. +func (s) TestCreateChannel_Errors(t *testing.T) { + c := &clientImpl{} + + tests := []struct { + name string + cfg *grpcservice.Config + wantErr string + }{ + { + name: "nil_config", + cfg: nil, + wantErr: "no channel credentials", + }, + { + name: "no_channel_creds", + cfg: &grpcservice.Config{TargetURI: "passthrough:///target"}, + wantErr: "no channel credentials", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, _, err := c.CreateChannel(tt.cfg) + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("CreateChannel() returned error %v, want error containing %q", err, tt.wantErr) + } + }) + } +} + +// Tests that a channel creation failure — here, call credentials requiring +// transport security combined with insecure channel credentials — fails +// CreateChannel and releases the config's owned credentials. +func (s) TestCreateChannel_DialError(t *testing.T) { + c := &clientImpl{} + + tokenCreds, err := accesstokencreds.NewCallCredentials(json.RawMessage(`{"token": "test-token"}`)) + if err != nil { + t.Fatalf("NewCallCredentials() failed: %v", err) + } + released := false + cfg := &grpcservice.Config{ + TargetURI: "passthrough:///target", + ChannelCredentials: testChannelCreds("insecure", func() { released = true }), + CallCredentials: []*creds.CallCreds{ + creds.NewCallCreds(tokenCreds, creds.NewJSONIdentity("access_token", nil), func() {}), + }, + } + if _, _, err := c.CreateChannel(cfg); err == nil || !strings.Contains(err.Error(), "transport level security") { + t.Fatalf("CreateChannel() returned error %v, want transport security error", err) + } + if !released { + t.Fatal("CreateChannel() did not release the config's credentials on failure") + } +} diff --git a/internal/xds/xdsclient/client.go b/internal/xds/xdsclient/client.go index d05382d0b112..0b197cbb92e1 100644 --- a/internal/xds/xdsclient/client.go +++ b/internal/xds/xdsclient/client.go @@ -23,9 +23,11 @@ package xdsclient import ( "context" + "google.golang.org/grpc" "google.golang.org/grpc/internal/xds/bootstrap" "google.golang.org/grpc/internal/xds/clients/lrsclient" "google.golang.org/grpc/internal/xds/clients/xdsclient" + "google.golang.org/grpc/internal/xds/grpcservice" v3statuspb "github.com/envoyproxy/go-control-plane/envoy/service/status/v3" ) @@ -53,6 +55,15 @@ type XDSClient interface { ReportLoad(*bootstrap.ServerConfig) (*lrsclient.LoadStore, func(context.Context)) BootstrapConfig() *bootstrap.Config + + // CreateChannel returns a shared gRPC channel to the side-channel service + // described by the given config, creating it on first use. A channel is + // shared between configs with the same target and credential identities. + // The returned release function must be called when the caller is done + // with the channel; the channel is closed when the last user releases it. + // Credentials owned by the config are released when the channel is + // closed, or immediately if an equivalent channel already exists. + CreateChannel(cfg *grpcservice.Config) (grpc.ClientConnInterface, func(), error) } // DumpResources returns the status and contents of all xDS resources. It uses diff --git a/internal/xds/xdsclient/clientimpl.go b/internal/xds/xdsclient/clientimpl.go index 23d6227300cd..7863d2fcfd46 100644 --- a/internal/xds/xdsclient/clientimpl.go +++ b/internal/xds/xdsclient/clientimpl.go @@ -20,6 +20,7 @@ package xdsclient import ( "fmt" + "sync" "sync/atomic" "time" @@ -107,6 +108,12 @@ type clientImpl struct { // Accessed atomically refCount int32 + + // Pool of shared side channels created via CreateChannel (gRFC A102). + // Lookups compare the entries' configs for equality; configs are never + // used as map keys. + sideChannelsMu sync.Mutex + sideChannels []*sideChannelEntry } // metricsReporter implements the clients.MetricsReporter interface and uses an diff --git a/internal/xds/xdsclient/pool.go b/internal/xds/xdsclient/pool.go index fa11a9500775..cb34a29fcd44 100644 --- a/internal/xds/xdsclient/pool.go +++ b/internal/xds/xdsclient/pool.go @@ -255,6 +255,15 @@ func (p *Pool) clientRefCountedClose(name string) { } } } + // The allowed-services credentials are owned by the (pool-shared) + // bootstrap config and are released when the last reference to the + // client is dropped, following the same lifecycle as the xDS server + // credentials released above. + for _, svc := range client.bootstrapConfig.AllowedGRPCServices() { + for _, f := range svc.Cleanups() { + f() + } + } p.mu.Unlock() // This attempts to close the transport to the management server and could diff --git a/internal/xds/xdsclient/xdsresource/grpc_service.go b/internal/xds/xdsclient/xdsresource/grpc_service.go deleted file mode 100644 index c4571d0900e0..000000000000 --- a/internal/xds/xdsclient/xdsresource/grpc_service.go +++ /dev/null @@ -1,49 +0,0 @@ -/* - * - * Copyright 2026 gRPC authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package xdsresource - -import ( - "time" - - "google.golang.org/grpc/metadata" -) - -// GRPCServiceConfig contains the configuration for an external server. It is -// the parsed configuration for the GrpcService proto message. -// See: https://www.envoyproxy.io/docs/envoy/latest/api-v3/config/core/v3/grpc_service.proto -type GRPCServiceConfig struct { - // TargetURI is the name of the external server. - TargetURI string - // ChannelCredentials specifies the configuration for the transport - // credentials to use to connect to the external server, as a JSON string. - ChannelCredentials string - // CallCredentials specifies the configuration for the per-RPC credentials to - // use when making calls to the external server, as a JSON string. - CallCredentials string - // Timeout is the RPC timeout for the call to the external server. If unset, - // the timeout depends on the usage of this external server. For example, - // cases like ext_authz and ext_proc, where there is a 1:1 mapping between the - // data plane RPC and the external server call, the timeout will be capped by - // the timeout on the data plane RPC. For cases like RLQS where there is a - // side channel to the external server, an unset timeout will result in no - // timeout being applied to the external server call. - Timeout time.Duration - // InitialMetadata is the additional metadata to include in all RPCs sent to - // the external server. - InitialMetadata metadata.MD -} diff --git a/xds/bootstrap/credentials.go b/xds/bootstrap/credentials.go index 254775942038..657d203e30a4 100644 --- a/xds/bootstrap/credentials.go +++ b/xds/bootstrap/credentials.go @@ -26,6 +26,7 @@ import ( "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/internal/xds/bootstrap/jwtcreds" "google.golang.org/grpc/internal/xds/bootstrap/tlscreds" + "google.golang.org/grpc/internal/xds/grpcservice/accesstokencreds" ) func init() { @@ -34,6 +35,7 @@ func init() { RegisterChannelCredentials(&tlsCredsBuilder{}) RegisterCallCredentials(&jwtCallCredsBuilder{}) + RegisterCallCredentials(&accessTokenCallCredsBuilder{}) } // insecureCredsBuilder implements the `ChannelCredentials` interface defined in @@ -83,3 +85,22 @@ func (j *jwtCallCredsBuilder) Build(configJSON json.RawMessage) (credentials.Per func (j *jwtCallCredsBuilder) Name() string { return "jwt_token_file" } + +// accessTokenCallCredsBuilder implements the `CallCredentials` interface +// defined in package `xds/bootstrap` and encapsulates static access token +// call credentials (gRFC A102). +type accessTokenCallCredsBuilder struct{} + +func (a *accessTokenCallCredsBuilder) Build(configJSON json.RawMessage) (credentials.PerRPCCredentials, func(), error) { + cc, err := accesstokencreds.NewCallCredentials(configJSON) + if err != nil { + return nil, nil, err + } + // These credentials hold no resources; the no-op cleanup satisfies the + // registry's Build contract. + return cc, func() {}, nil +} + +func (a *accessTokenCallCredsBuilder) Name() string { + return "access_token" +}