/* * * Copyright 2021 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 ( "errors" "fmt" "net/netip" "strconv" v1xdsudpatypepb "github.com/cncf/xds/go/udpa/type/v1" v3xdsxdstypepb "github.com/cncf/xds/go/xds/type/v3" v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3" v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3" v3httppb "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/http_connection_manager/v3" v3tlspb "github.com/envoyproxy/go-control-plane/envoy/extensions/transport_sockets/tls/v3" "google.golang.org/grpc/internal/envconfig" "google.golang.org/grpc/internal/xds/bootstrap" "google.golang.org/grpc/internal/xds/httpfilter" "google.golang.org/grpc/internal/xds/xdsclient/xdsresource/version" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/anypb" ) func unmarshalListenerResource(r *anypb.Any, bc *bootstrap.Config, sc *bootstrap.ServerConfig) (string, ListenerUpdate, error) { r, err := UnwrapResource(r) if err != nil { return "", ListenerUpdate{}, fmt.Errorf("failed to unwrap resource: %v", err) } if !IsListenerResource(r.GetTypeUrl()) { return "", ListenerUpdate{}, fmt.Errorf("unexpected listener resource type: %q ", r.GetTypeUrl()) } lis := &v3listenerpb.Listener{} if err := proto.Unmarshal(r.GetValue(), lis); err != nil { return "", ListenerUpdate{}, fmt.Errorf("failed to unmarshal resource: %v", err) } if lis.GetName() == "" { return "", ListenerUpdate{}, fmt.Errorf("empty resource name in listener resource") } lu, err := processListener(lis, bc, sc) if err != nil { return lis.GetName(), ListenerUpdate{}, err } lu.Raw = r return lis.GetName(), *lu, nil } func processListener(lis *v3listenerpb.Listener, bc *bootstrap.Config, sc *bootstrap.ServerConfig) (*ListenerUpdate, error) { if lis.GetApiListener() != nil { return processClientSideListener(lis, bc, sc) } return processServerSideListener(lis, bc, sc) } // processClientSideListener checks if the provided Listener proto meets // the expected criteria. If so, it returns a non-empty routeConfigName. func processClientSideListener(lis *v3listenerpb.Listener, bc *bootstrap.Config, sc *bootstrap.ServerConfig) (*ListenerUpdate, error) { apiLisAny := lis.GetApiListener().GetApiListener() if !IsHTTPConnManagerResource(apiLisAny.GetTypeUrl()) { return nil, fmt.Errorf("unexpected http connection manager resource type: %q", apiLisAny.GetTypeUrl()) } apiLis := &v3httppb.HttpConnectionManager{} if err := proto.Unmarshal(apiLisAny.GetValue(), apiLis); err != nil { return nil, fmt.Errorf("failed to unmarshal api_listener: %v", err) } // "HttpConnectionManager.xff_num_trusted_hops must be unset or zero and // HttpConnectionManager.original_ip_detection_extensions must be empty. If // either field has an incorrect value, the Listener must be NACKed." - A41 if apiLis.XffNumTrustedHops != 0 { return nil, fmt.Errorf("xff_num_trusted_hops must be unset or zero %+v", apiLis) } if len(apiLis.OriginalIpDetectionExtensions) != 0 { return nil, fmt.Errorf("original_ip_detection_extensions must be empty %+v", apiLis) } hcm := &HTTPConnectionManagerConfig{} switch apiLis.RouteSpecifier.(type) { case *v3httppb.HttpConnectionManager_Rds: if configsource := apiLis.GetRds().GetConfigSource(); configsource.GetAds() == nil && configsource.GetSelf() == nil { return nil, fmt.Errorf("LDS's RDS configSource is not ADS or Self: %+v", lis) } name := apiLis.GetRds().GetRouteConfigName() if name == "" { return nil, fmt.Errorf("empty route_config_name: %+v", lis) } hcm.RouteConfigName = name case *v3httppb.HttpConnectionManager_RouteConfig: routeU, err := generateRDSUpdateFromRouteConfiguration(apiLis.GetRouteConfig(), bc, sc) if err != nil { return nil, fmt.Errorf("failed to parse inline RDS resp: %v", err) } hcm.InlineRouteConfig = &routeU case nil: return nil, fmt.Errorf("no RouteSpecifier: %+v", apiLis) default: return nil, fmt.Errorf("unsupported type %T for RouteSpecifier", apiLis.RouteSpecifier) } // The following checks and fields only apply to xDS protocol versions v3+. hcm.MaxStreamDuration = apiLis.GetCommonHttpProtocolOptions().GetMaxStreamDuration().AsDuration() var err error if hcm.HTTPFilters, err = processHTTPFilters(apiLis.GetHttpFilters(), false, bc, sc); err != nil { return nil, err } return &ListenerUpdate{APIListener: hcm}, nil } func unwrapHTTPFilterConfig(config *anypb.Any) (proto.Message, string, error) { switch { case config.MessageIs(&v3xdsxdstypepb.TypedStruct{}): // The real type name is inside the new TypedStruct message. s := new(v3xdsxdstypepb.TypedStruct) if err := config.UnmarshalTo(s); err != nil { return nil, "", fmt.Errorf("error unmarshalling TypedStruct filter config: %v", err) } return s, s.GetTypeUrl(), nil case config.MessageIs(&v1xdsudpatypepb.TypedStruct{}): // The real type name is inside the old TypedStruct message. s := new(v1xdsudpatypepb.TypedStruct) if err := config.UnmarshalTo(s); err != nil { return nil, "", fmt.Errorf("error unmarshalling TypedStruct filter config: %v", err) } return s, s.GetTypeUrl(), nil default: return config, config.GetTypeUrl(), nil } } func validateHTTPFilterConfig(cfg *anypb.Any, lds, optional bool, bc *bootstrap.Config, sc *bootstrap.ServerConfig) (httpfilter.Builder, httpfilter.FilterConfig, error) { config, typeURL, err := unwrapHTTPFilterConfig(cfg) if err != nil { return nil, nil, err } filterBuilder := httpfilter.Get(typeURL) if filterBuilder == nil { if optional { return nil, nil, nil } return nil, nil, fmt.Errorf("no filter implementation found for %q", typeURL) } parseFunc := filterBuilder.ParseFilterConfig if !lds { parseFunc = filterBuilder.ParseFilterConfigOverride } filterConfig, err := parseFunc(config, httpfilter.ParseOptions{BootstrapConfig: bc, ServerConfig: sc}) if err != nil { return nil, nil, fmt.Errorf("error parsing config for filter %q: %v", typeURL, err) } return filterBuilder, filterConfig, nil } func processHTTPFilterOverrides(cfgs map[string]*anypb.Any, bc *bootstrap.Config, sc *bootstrap.ServerConfig) (map[string]httpfilter.FilterConfig, error) { if len(cfgs) == 0 { return nil, nil } m := make(map[string]httpfilter.FilterConfig) for name, cfg := range cfgs { optional := false disabled := false s := new(v3routepb.FilterConfig) if cfg.MessageIs(s) { if err := cfg.UnmarshalTo(s); err != nil { return nil, fmt.Errorf("filter override %q: error unmarshalling FilterConfig: %v", name, err) } cfg = s.GetConfig() optional = s.GetIsOptional() if envconfig.XDSClientExtProcEnabled || envconfig.XDSClientExtAuthzEnabled { disabled = s.GetDisabled() } } if disabled { m[name] = httpfilter.DisabledFilterConfig{} continue } httpFilter, config, err := validateHTTPFilterConfig(cfg, false, optional, bc, sc) if err != nil { return nil, fmt.Errorf("filter override %q: %v", name, err) } if httpFilter == nil { // Optional configs are ignored. continue } m[name] = config } return m, nil } func processHTTPFilters(filters []*v3httppb.HttpFilter, server bool, bc *bootstrap.Config, sc *bootstrap.ServerConfig) ([]HTTPFilter, error) { ret := make([]HTTPFilter, 0, len(filters)) seenNames := make(map[string]bool, len(filters)) for _, filter := range filters { name := filter.GetName() if name == "" { return nil, errors.New("filter missing name field") } if seenNames[name] { return nil, fmt.Errorf("duplicate filter name %q", name) } seenNames[name] = true httpFilter, config, err := validateHTTPFilterConfig(filter.GetTypedConfig(), true, filter.GetIsOptional(), bc, sc) if err != nil { return nil, err } if httpFilter == nil { // Optional configs are ignored. continue } if server { if _, ok := httpFilter.(httpfilter.ServerFilterBuilder); !ok { if filter.GetIsOptional() { continue } return nil, fmt.Errorf("HTTP filter %q not supported server-side", name) } } else if _, ok := httpFilter.(httpfilter.ClientFilterBuilder); !ok { if filter.GetIsOptional() { continue } return nil, fmt.Errorf("HTTP filter %q not supported client-side", name) } disabled := false if envconfig.XDSClientExtProcEnabled || envconfig.XDSClientExtAuthzEnabled { disabled = filter.GetDisabled() } // Save name/config ret = append(ret, HTTPFilter{Name: name, Filter: httpFilter, Config: config, Disabled: disabled}) } // "Validation will fail if a terminal filter is not the last filter in the // chain or if a non-terminal filter is the last filter in the chain." - A39 if len(ret) == 0 { return nil, fmt.Errorf("http filters list is empty") } var i int for ; i < len(ret)-1; i++ { if ret[i].Filter.IsTerminal() { return nil, fmt.Errorf("http filter %q is a terminal filter but it is not last in the filter chain", ret[i].Name) } } if !ret[i].Filter.IsTerminal() { return nil, fmt.Errorf("http filter %q is not a terminal filter", ret[len(ret)-1].Name) } return ret, nil } func processServerSideListener(lis *v3listenerpb.Listener, bc *bootstrap.Config, sc *bootstrap.ServerConfig) (*ListenerUpdate, error) { if n := len(lis.ListenerFilters); n != 0 { return nil, fmt.Errorf("unsupported field 'listener_filters' contains %d entries", n) } if lis.GetUseOriginalDst().GetValue() { return nil, errors.New("unsupported field 'use_original_dst' is present and set to true") } addr := lis.GetAddress() if addr == nil { return nil, fmt.Errorf("no address field in LDS response: %+v", lis) } sockAddr := addr.GetSocketAddress() if sockAddr == nil { return nil, fmt.Errorf("no socket_address field in LDS response: %+v", lis) } lu := &ListenerUpdate{ TCPListener: &InboundListenerConfig{ Address: sockAddr.GetAddress(), Port: strconv.Itoa(int(sockAddr.GetPortValue())), }, } // Populate the default filter chain. if dfc := lis.GetDefaultFilterChain(); dfc != nil { fc, err := filterChainFromProto(dfc, bc, sc) if err != nil { return nil, fmt.Errorf("failed to unmarshal default filter chain: %v", err) } lu.TCPListener.DefaultFilterChain = fc } // Populated the filter chain map. fcm, err := buildFilterChainMap(lis.GetFilterChains(), bc, sc) if err != nil { return nil, fmt.Errorf("failed to unmarshal filter chains: %v", err) } lu.TCPListener.FilterChains = fcm // If there are no supported filter chains and no default filter chain, we // fail here. This will cause the Listener resource to be NACK'ed. if len(lu.TCPListener.FilterChains.DstPrefixes) == 0 && lu.TCPListener.DefaultFilterChain.IsEmpty() { return nil, fmt.Errorf("no supported filter chains and no default filter chain") } return lu, nil } func filterChainFromProto(fc *v3listenerpb.FilterChain, bc *bootstrap.Config, sc *bootstrap.ServerConfig) (NetworkFilterChainConfig, error) { var emptyFilterChain NetworkFilterChainConfig hcmConfig, err := processNetworkFilters(fc.GetFilters(), bc, sc) if err != nil { return emptyFilterChain, err } fcc := NetworkFilterChainConfig{HTTPConnMgr: hcmConfig} // If the transport_socket field is not specified, it means that the control // plane has not sent us any security config. This is fine and the server // will use the fallback credentials configured as part of the // xdsCredentials. ts := fc.GetTransportSocket() if ts == nil { return fcc, nil } if name := ts.GetName(); name != transportSocketName { return emptyFilterChain, fmt.Errorf("transport_socket field has unexpected name: %s", name) } tc := ts.GetTypedConfig() if typeURL := tc.GetTypeUrl(); typeURL != version.V3DownstreamTLSContextURL { return emptyFilterChain, fmt.Errorf("transport_socket missing typed_config or wrong type_url: %q", typeURL) } downstreamCtx := &v3tlspb.DownstreamTlsContext{} if err := proto.Unmarshal(tc.GetValue(), downstreamCtx); err != nil { return emptyFilterChain, fmt.Errorf("failed to unmarshal DownstreamTlsContext in LDS response: %v", err) } if downstreamCtx.GetRequireSni().GetValue() { return emptyFilterChain, fmt.Errorf("require_sni field set to true in DownstreamTlsContext message: %v", downstreamCtx) } if downstreamCtx.GetOcspStaplePolicy() != v3tlspb.DownstreamTlsContext_LENIENT_STAPLING { return emptyFilterChain, fmt.Errorf("ocsp_staple_policy field set to unsupported value in DownstreamTlsContext message: %v", downstreamCtx) } if downstreamCtx.GetCommonTlsContext() == nil { return emptyFilterChain, errors.New("DownstreamTlsContext in LDS response does not contain a CommonTlsContext") } secCfg, err := securityConfigFromCommonTLSContext(downstreamCtx.GetCommonTlsContext(), true) if err != nil { return emptyFilterChain, err } if secCfg != nil { secCfg.RequireClientCert = downstreamCtx.GetRequireClientCertificate().GetValue() if secCfg.RequireClientCert && secCfg.RootInstanceName == "" { return emptyFilterChain, errors.New("security configuration on the server-side does not contain root certificate provider instance name, but require_client_cert field is set") } fcc.SecurityCfg = secCfg } return fcc, nil } // dstPrefixEntry wraps DestinationPrefixEntry to track build state. type dstPrefixEntry struct { entry DestinationPrefixEntry rawBufferSeen bool } func buildFilterChainMap(fcs []*v3listenerpb.FilterChain, bc *bootstrap.Config, sc *bootstrap.ServerConfig) (NetworkFilterChainMap, error) { dstPrefixEntries := []*dstPrefixEntry{} for _, fc := range fcs { fcMatch := fc.GetFilterChainMatch() if fcMatch.GetDestinationPort().GetValue() != 0 { // Destination port is the first match criteria and we do not // support filter chains that contain this match criteria. logger.Warningf("Dropping filter chain %q since it contains unsupported destination_port match field", fc.GetName()) continue } var err error dstPrefixEntries, err = addFilterChainsForDestPrefixes(dstPrefixEntries, fc, bc, sc) if err != nil { return NetworkFilterChainMap{}, err } } entries := []DestinationPrefixEntry{} for _, bEntry := range dstPrefixEntries { fcSeen := false for _, srcPrefixes := range bEntry.entry.SourceTypeArr { if len(srcPrefixes.Entries) == 0 { continue } for _, srcPrefix := range srcPrefixes.Entries { for _, fc := range srcPrefix.PortMap { if !fc.IsEmpty() { fcSeen = true } } } } if fcSeen { entries = append(entries, bEntry.entry) } } return NetworkFilterChainMap{DstPrefixes: entries}, nil } func addFilterChainsForDestPrefixes(dstPrefixEntries []*dstPrefixEntry, fc *v3listenerpb.FilterChain, bc *bootstrap.Config, sc *bootstrap.ServerConfig) ([]*dstPrefixEntry, error) { ranges := fc.GetFilterChainMatch().GetPrefixRanges() dstPrefixes, err := parsePrefixRanges(ranges) if err != nil { return nil, fmt.Errorf("failed to parse destination prefix ranges: %v", err) } var entry *dstPrefixEntry if len(dstPrefixes) == 0 { // Use the unspecified entry when destination prefix is unspecified, and // set the `prefix` field to nil. dstPrefixEntries, entry = getOrCreateDestPrefixEntry(dstPrefixEntries, netip.Prefix{}) if err := addFilterChainsForServerNames(entry, fc, bc, sc); err != nil { return nil, err } return dstPrefixEntries, nil } for _, prefix := range dstPrefixes { dstPrefixEntries, entry = getOrCreateDestPrefixEntry(dstPrefixEntries, prefix) if err := addFilterChainsForServerNames(entry, fc, bc, sc); err != nil { return nil, err } } return dstPrefixEntries, nil } // getOrCreateDestPrefixEntry looks for an existing dstPrefixEntry in the // provided slice with the same destination prefix as the provided prefix. If // such an entry is found, it is returned. Otherwise, a new entry is created and // appended to the slice, and the new entry is returned. func getOrCreateDestPrefixEntry(dstPrefixEntries []*dstPrefixEntry, prefix netip.Prefix) ([]*dstPrefixEntry, *dstPrefixEntry) { for _, e := range dstPrefixEntries { if e.entry.Prefix == prefix { return dstPrefixEntries, e } } entry := &dstPrefixEntry{entry: DestinationPrefixEntry{Prefix: prefix}} dstPrefixEntries = append(dstPrefixEntries, entry) return dstPrefixEntries, entry } func addFilterChainsForServerNames(dstEntry *dstPrefixEntry, fc *v3listenerpb.FilterChain, bc *bootstrap.Config, sc *bootstrap.ServerConfig) error { // Filter chains specifying server names in their match criteria always fail // a match at connection time. So, these filter chains can be dropped now. if len(fc.GetFilterChainMatch().GetServerNames()) != 0 { logger.Warningf("Dropping filter chain %q since it contains unsupported server_names match field", fc.GetName()) return nil } return addFilterChainsForTransportProtocols(dstEntry, fc, bc, sc) } func addFilterChainsForTransportProtocols(dstEntry *dstPrefixEntry, fc *v3listenerpb.FilterChain, bc *bootstrap.Config, sc *bootstrap.ServerConfig) error { tp := fc.GetFilterChainMatch().GetTransportProtocol() switch { case tp != "" && tp != "raw_buffer": // Only allow filter chains with transport protocol set to empty string // or "raw_buffer". logger.Warningf("Dropping filter chain %q since it contains unsupported value for transport_protocol match field", fc.GetName()) return nil case tp == "" && dstEntry.rawBufferSeen: // If we have already seen filter chains with transport protocol set to // "raw_buffer", we can drop filter chains with transport protocol set // to empty string, since the former takes precedence. logger.Warningf("Dropping filter chain %q since it contains empty string for transport_protocol match field, but one with raw_buffer already exists", fc.GetName()) return nil case tp != "" && !dstEntry.rawBufferSeen: // This is the first "raw_buffer" that we are seeing. Set the bit and // reset the source types array which might contain entries for filter // chains with transport protocol set to empty string. dstEntry.rawBufferSeen = true dstEntry.entry.SourceTypeArr = [3]SourcePrefixes{} } return addFilterChainsForApplicationProtocols(dstEntry, fc, bc, sc) } func addFilterChainsForApplicationProtocols(dstEntry *dstPrefixEntry, fc *v3listenerpb.FilterChain, bc *bootstrap.Config, sc *bootstrap.ServerConfig) error { if len(fc.GetFilterChainMatch().GetApplicationProtocols()) != 0 { logger.Warningf("Dropping filter chain %q since it contains unsupported application_protocols match field", fc.GetName()) return nil } return addFilterChainsForSourceType(&dstEntry.entry, fc, bc, sc) } // sourceType specifies the connection source IP match type. type sourceType int const ( // sourceTypeAny matches connection attempts from any source. sourceTypeAny sourceType = iota // sourceTypeSameOrLoopback matches connection attempts from the same host. sourceTypeSameOrLoopback // sourceTypeExternal matches connection attempts from a different host. sourceTypeExternal ) func addFilterChainsForSourceType(entry *DestinationPrefixEntry, fc *v3listenerpb.FilterChain, bc *bootstrap.Config, sc *bootstrap.ServerConfig) error { var srcType sourceType switch st := fc.GetFilterChainMatch().GetSourceType(); st { case v3listenerpb.FilterChainMatch_ANY: srcType = sourceTypeAny case v3listenerpb.FilterChainMatch_SAME_IP_OR_LOOPBACK: srcType = sourceTypeSameOrLoopback case v3listenerpb.FilterChainMatch_EXTERNAL: srcType = sourceTypeExternal default: return fmt.Errorf("unsupported source type: %v", st) } return addFilterChainsForSourcePrefixes(&entry.SourceTypeArr[srcType], fc, bc, sc) } func addFilterChainsForSourcePrefixes(srcPrefixes *SourcePrefixes, fc *v3listenerpb.FilterChain, bc *bootstrap.Config, sc *bootstrap.ServerConfig) error { ranges := fc.GetFilterChainMatch().GetSourcePrefixRanges() prefixes, err := parsePrefixRanges(ranges) if err != nil { return fmt.Errorf("failed to parse source prefix ranges: %v", err) } if len(prefixes) == 0 { return getOrCreateSourcePrefixEntry(srcPrefixes, netip.Prefix{}, fc, bc, sc) } for _, prefix := range prefixes { if err := getOrCreateSourcePrefixEntry(srcPrefixes, prefix, fc, bc, sc); err != nil { return err } } return nil } func parsePrefixRanges(ranges []*v3corepb.CidrRange) ([]netip.Prefix, error) { prefixes := make([]netip.Prefix, 0, len(ranges)) for _, pr := range ranges { addrStr := pr.GetAddressPrefix() bits := int(pr.GetPrefixLen().GetValue()) addr, err := netip.ParseAddr(addrStr) if err != nil { return nil, fmt.Errorf("invalid address: %q", addrStr) } prefix := netip.PrefixFrom(addr.Unmap(), bits).Masked() if !prefix.IsValid() { return nil, fmt.Errorf(`length %d is invalid for "%s" (max %d)`, bits, addrStr, addr.BitLen()) } prefixes = append(prefixes, prefix) } return prefixes, nil } // getOrCreateSourcePrefixEntry looks for an existing SourcePrefixEntry in the // provided SourcePrefixes with the same source prefix as the provided prefix. If // such an entry is found, the provided filter chain is added to the entry and // nil is returned. Otherwise, a new entry is created and appended to the // SourcePrefixes, the provided filter chain is added to the new entry, and nil // is returned. If there are multiple filter chains with overlapping matching // rules, an error is returned. func getOrCreateSourcePrefixEntry(srcPrefixes *SourcePrefixes, prefix netip.Prefix, fc *v3listenerpb.FilterChain, bc *bootstrap.Config, sc *bootstrap.ServerConfig) error { for i := range srcPrefixes.Entries { if srcPrefixes.Entries[i].Prefix == prefix { return addFilterChainsForSourcePorts(&srcPrefixes.Entries[i], fc, bc, sc) } } // Not found, create a new entry. srcPrefixes.Entries = append(srcPrefixes.Entries, SourcePrefixEntry{ Prefix: prefix, PortMap: make(map[int]NetworkFilterChainConfig), }) return addFilterChainsForSourcePorts(&srcPrefixes.Entries[len(srcPrefixes.Entries)-1], fc, bc, sc) } func addFilterChainsForSourcePorts(entry *SourcePrefixEntry, fc *v3listenerpb.FilterChain, bc *bootstrap.Config, sc *bootstrap.ServerConfig) error { ports := fc.GetFilterChainMatch().GetSourcePorts() srcPorts := make([]int, 0, len(ports)) for _, port := range ports { srcPorts = append(srcPorts, int(port)) } if len(srcPorts) == 0 { if !entry.PortMap[0].IsEmpty() { return errors.New("multiple filter chains with overlapping matching rules are defined") } fcc, err := filterChainFromProto(fc, bc, sc) if err != nil { return err } entry.PortMap[0] = fcc return nil } for _, port := range srcPorts { if !entry.PortMap[port].IsEmpty() { return errors.New("multiple filter chains with overlapping matching rules are defined") } fcc, err := filterChainFromProto(fc, bc, sc) if err != nil { return err } entry.PortMap[port] = fcc } return nil }