#19717

/*
 * Copyright 2021 The 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 io.grpc.xds;

import static com.google.common.base.Preconditions.checkNotNull;
import static io.grpc.xds.XdsAttributes.ATTR_DRAIN_GRACE_NANOS;
import static io.grpc.xds.XdsAttributes.ATTR_FILTER_CHAIN_SELECTOR_MANAGER;
import static io.grpc.xds.XdsServerWrapper.ATTR_SERVER_ROUTING_CONFIG;
import static io.grpc.xds.internal.security.SecurityProtocolNegotiators.ATTR_SERVER_SSL_CONTEXT_PROVIDER_SUPPLIER;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import com.google.common.base.Strings;
import com.google.common.collect.Iterables;
import com.google.protobuf.UInt32Value;
import io.grpc.Attributes;
import io.grpc.internal.ObjectPool;
import io.grpc.netty.GrpcHttp2ConnectionHandler;
import io.grpc.netty.InternalGracefulServerCloseCommand;
import io.grpc.netty.InternalProtocolNegotiationEvent;
import io.grpc.netty.InternalProtocolNegotiator;
import io.grpc.netty.InternalProtocolNegotiator.ProtocolNegotiator;
import io.grpc.netty.ProtocolNegotiationEvent;
import io.grpc.xds.EnvoyServerProtoData.CidrRange;
import io.grpc.xds.EnvoyServerProtoData.ConnectionSourceType;
import io.grpc.xds.EnvoyServerProtoData.FilterChain;
import io.grpc.xds.EnvoyServerProtoData.FilterChainMatch;
import io.grpc.xds.XdsServerWrapper.ServerRoutingConfig;
import io.grpc.xds.internal.Matchers.CidrMatcher;
import io.grpc.xds.internal.security.SslContextProviderSupplier;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.AsciiString;
import java.net.Inet6Address;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.Nullable;


/**
 * Handles L4 filter chain match for the connection based on the xds configuration.
 * */
@SuppressWarnings("FutureReturnValueIgnored") // Netty doesn't follow this pattern
final class FilterChainMatchingProtocolNegotiators {
  private static final Logger log = Logger.getLogger(
          FilterChainMatchingProtocolNegotiators.class.getName());

  private static final AsciiString SCHEME = AsciiString.of("http");

  private FilterChainMatchingProtocolNegotiators() {
  }

  @VisibleForTesting
  static final class FilterChainMatchingHandler extends ChannelInboundHandlerAdapter {

    private final GrpcHttp2ConnectionHandler grpcHandler;
    private final FilterChainSelectorManager filterChainSelectorManager;
    private final ProtocolNegotiator delegate;

    FilterChainMatchingHandler(
            GrpcHttp2ConnectionHandler grpcHandler,
            FilterChainSelectorManager filterChainSelectorManager,
            ProtocolNegotiator delegate) {
      this.grpcHandler = checkNotNull(grpcHandler, "grpcHandler");
      this.filterChainSelectorManager =
          checkNotNull(filterChainSelectorManager, "filterChainSelectorManager");
      this.delegate = checkNotNull(delegate, "delegate");
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
      if (!(evt instanceof ProtocolNegotiationEvent)) {
        super.userEventTriggered(ctx, evt);
        return;
      }
      long drainGraceTime = 0;
      TimeUnit drainGraceTimeUnit = null;
      Long drainGraceNanosObj = grpcHandler.getEagAttributes().get(ATTR_DRAIN_GRACE_NANOS);
      if (drainGraceNanosObj != null) {
        drainGraceTime = drainGraceNanosObj;
        drainGraceTimeUnit = TimeUnit.NANOSECONDS;
      }
      FilterChainSelectorManager.Closer closer = new FilterChainSelectorManager.Closer(
          new GracefullyShutdownChannelRunnable(ctx.channel(), drainGraceTime, drainGraceTimeUnit));
      FilterChainSelector selector = filterChainSelectorManager.register(closer);
      ctx.channel().closeFuture().addListener(
          new FilterChainSelectorManagerDeregister(filterChainSelectorManager, closer));
      checkNotNull(selector, "selector");
      SelectedConfig config = selector.select(
          (InetSocketAddress) ctx.channel().localAddress(),
          (InetSocketAddress) ctx.channel().remoteAddress());
      if (config == null) {
        log.log(Level.WARNING, "Connection from {0} to {1} has no matching filter chain. Closing",
            new Object[] {ctx.channel().remoteAddress(), ctx.channel().localAddress()});
        ctx.close().addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
        return;
      }
      ProtocolNegotiationEvent pne = (ProtocolNegotiationEvent) evt;
      // TODO(zivy): merge into one key and take care of this outer class visibility.
      Attributes attr = InternalProtocolNegotiationEvent.getAttributes(pne).toBuilder()
              .set(ATTR_SERVER_SSL_CONTEXT_PROVIDER_SUPPLIER, config.sslContextProviderSupplier)
              .set(ATTR_SERVER_ROUTING_CONFIG, config.routingConfig)
              .build();
      pne = InternalProtocolNegotiationEvent.withAttributes(pne, attr);
      ctx.pipeline().replace(this, null, delegate.newHandler(grpcHandler));
      ctx.fireUserEventTriggered(pne);
    }

    static final class FilterChainSelector {
      public static final FilterChainSelector NO_FILTER_CHAIN = new FilterChainSelector(
              Collections.<FilterChain, AtomicReference<ServerRoutingConfig>>emptyMap(),
          null, new AtomicReference<ServerRoutingConfig>());
      private final Map<FilterChain, AtomicReference<ServerRoutingConfig>> routingConfigs;
      @Nullable
      private final SslContextProviderSupplier defaultSslContextProviderSupplier;
      private final AtomicReference<ServerRoutingConfig> defaultRoutingConfig;

      FilterChainSelector(Map<FilterChain, AtomicReference<ServerRoutingConfig>> routingConfigs,
                          @Nullable SslContextProviderSupplier defaultSslContextProviderSupplier,
                          AtomicReference<ServerRoutingConfig> defaultRoutingConfig) {
        this.routingConfigs = checkNotNull(routingConfigs, "routingConfigs");
        this.defaultSslContextProviderSupplier = defaultSslContextProviderSupplier;
        this.defaultRoutingConfig = checkNotNull(defaultRoutingConfig, "defaultRoutingConfig");
      }

      FilterChainSelector(Map<FilterChain, AtomicReference<ServerRoutingConfig>> routingConfigs) {
        this(routingConfigs, null, new AtomicReference<>());
      }

      @VisibleForTesting
      Map<FilterChain, AtomicReference<ServerRoutingConfig>> getRoutingConfigs() {
        return routingConfigs;
      }

      @VisibleForTesting
      AtomicReference<ServerRoutingConfig> getDefaultRoutingConfig() {
        return defaultRoutingConfig;
      }

      @VisibleForTesting
      SslContextProviderSupplier getDefaultSslContextProviderSupplier() {
        return defaultSslContextProviderSupplier;
      }

      /**
       * Throws IllegalStateException when no exact one match, and we should close the connection.
       */
      SelectedConfig select(InetSocketAddress localAddr, InetSocketAddress remoteAddr) {
        Collection<FilterChain> filterChains = routingConfigs.keySet();
        filterChains = filterOnDestinationPort(filterChains);
        filterChains = filterOnIpAddress(filterChains, localAddr.getAddress(), true);
        filterChains = filterOnServerNames(filterChains);
        filterChains = filterOnTransportProtocol(filterChains);
        filterChains = filterOnApplicationProtocols(filterChains);
        filterChains =
                filterOnSourceType(filterChains, remoteAddr.getAddress(), localAddr.getAddress());
        filterChains = filterOnIpAddress(filterChains, remoteAddr.getAddress(), false);
        filterChains = filterOnSourcePort(filterChains, remoteAddr.getPort());

        if (filterChains.size() > 1) {
          throw new IllegalStateException("Found more than one matching filter chains. This should "
              + "not be possible as ClientXdsClient validated the chains for uniqueness.");
        }
        if (filterChains.size() == 1) {
          FilterChain selected = Iterables.getOnlyElement(filterChains);
          return new SelectedConfig(
                  routingConfigs.get(selected), selected.sslContextProviderSupplier());
        }
        if (defaultRoutingConfig.get() != null) {
          return new SelectedConfig(defaultRoutingConfig, defaultSslContextProviderSupplier);
        }
        return null;
      }

      // reject if filer-chain-match has non-empty application_protocols
      private static Collection<FilterChain> filterOnApplicationProtocols(
              Collection<FilterChain> filterChains) {
        ArrayList<FilterChain> filtered = new ArrayList<>(filterChains.size());
        for (FilterChain filterChain : filterChains) {
          FilterChainMatch filterChainMatch = filterChain.filterChainMatch();

          if (filterChainMatch.applicationProtocols().isEmpty()) {
            filtered.add(filterChain);
          }
        }
        return filtered;
      }

      // reject if filer-chain-match has non-empty transport protocol other than "raw_buffer"
      private static Collection<FilterChain> filterOnTransportProtocol(
              Collection<FilterChain> filterChains) {
        ArrayList<FilterChain> filtered = new ArrayList<>(filterChains.size());
        for (FilterChain filterChain : filterChains) {
          FilterChainMatch filterChainMatch = filterChain.filterChainMatch();

          String transportProtocol = filterChainMatch.transportProtocol();
          if (Strings.isNullOrEmpty(transportProtocol) || "raw_buffer".equals(transportProtocol)) {
            filtered.add(filterChain);
          }
        }
        return filtered;
      }

      // reject if filer-chain-match has server_name(s)
      private static Collection<FilterChain> filterOnServerNames(
              Collection<FilterChain> filterChains) {
        ArrayList<FilterChain> filtered = new ArrayList<>(filterChains.size());
        for (FilterChain filterChain : filterChains) {
          FilterChainMatch filterChainMatch = filterChain.filterChainMatch();

          if (filterChainMatch.serverNames().isEmpty()) {
            filtered.add(filterChain);
          }
        }
        return filtered;
      }

      // destination_port present => Always fail match
      private static Collection<FilterChain> filterOnDestinationPort(
              Collection<FilterChain> filterChains) {
        ArrayList<FilterChain> filtered = new ArrayList<>(filterChains.size());
        for (FilterChain filterChain : filterChains) {
          FilterChainMatch filterChainMatch = filterChain.filterChainMatch();

          if (filterChainMatch.destinationPort()
                  == UInt32Value.getDefaultInstance().getValue()) {
            filtered.add(filterChain);
          }
        }
        return filtered;
      }

      private static Collection<FilterChain> filterOnSourcePort(
              Collection<FilterChain> filterChains, int sourcePort) {
        ArrayList<FilterChain> filteredOnMatch = new ArrayList<>(filterChains.size());
        ArrayList<FilterChain> filteredOnEmpty = new ArrayList<>(filterChains.size());
        for (FilterChain filterChain : filterChains) {
          FilterChainMatch filterChainMatch = filterChain.filterChainMatch();

          List<Integer> sourcePortsToMatch = filterChainMatch.sourcePorts();
          if (sourcePortsToMatch.isEmpty()) {
            filteredOnEmpty.add(filterChain);
          } else if (sourcePortsToMatch.contains(sourcePort)) {
            filteredOnMatch.add(filterChain);
          }
        }
        // match against source port is more specific than match against empty list
        return filteredOnMatch.isEmpty() ? filteredOnEmpty : filteredOnMatch;
      }

      private static Collection<FilterChain> filterOnSourceType(
              Collection<FilterChain> filterChains, InetAddress sourceAddress,
              InetAddress destAddress) {
        ArrayList<FilterChain> filtered = new ArrayList<>(filterChains.size());
        for (FilterChain filterChain : filterChains) {
          FilterChainMatch filterChainMatch = filterChain.filterChainMatch();
          ConnectionSourceType sourceType =
                  filterChainMatch.connectionSourceType();

          boolean matching = false;
          if (sourceType == ConnectionSourceType.SAME_IP_OR_LOOPBACK) {
            matching =
                    sourceAddress.isLoopbackAddress()
                            || sourceAddress.isAnyLocalAddress()
                            || sourceAddress.equals(destAddress);
          } else if (sourceType == ConnectionSourceType.EXTERNAL) {
            matching = !sourceAddress.isLoopbackAddress() && !sourceAddress.isAnyLocalAddress();
          } else { // ANY or null
            matching = true;
          }
          if (matching) {
            filtered.add(filterChain);
          }
        }
        return filtered;
      }

      private static int getMatchingPrefixLength(
              FilterChainMatch filterChainMatch, InetAddress address, boolean forDestination) {
        boolean isIPv6 = address instanceof Inet6Address;
        List<CidrRange> cidrRanges =
                forDestination
                        ? filterChainMatch.prefixRanges()
                        : filterChainMatch.sourcePrefixRanges();
        int matchingPrefixLength;
        if (cidrRanges.isEmpty()) { // if there is no CidrRange assume 0-length match
          matchingPrefixLength = 0;
        } else {
          matchingPrefixLength = -1;
          for (CidrRange cidrRange : cidrRanges) {
            InetAddress cidrAddr = cidrRange.addressPrefix();
            boolean cidrIsIpv6 = cidrAddr instanceof Inet6Address;
            if (isIPv6 == cidrIsIpv6) {
              int prefixLen = cidrRange.prefixLen();
              CidrMatcher matcher = CidrMatcher.create(cidrAddr, prefixLen);
              if (matcher.matches(address) && prefixLen > matchingPrefixLength) {
                matchingPrefixLength = prefixLen;
              }
            }
          }
        }
        return matchingPrefixLength;
      }

      // use prefix_ranges (CIDR) and get the most specific matches
      private static Collection<FilterChain> filterOnIpAddress(
              Collection<FilterChain> filterChains, InetAddress address, boolean forDestination) {
        // current list of top ones
        ArrayList<FilterChain> topOnes = new ArrayList<>(filterChains.size());
        int topMatchingPrefixLen = -1;
        for (FilterChain filterChain : filterChains) {
          int currentMatchingPrefixLen = getMatchingPrefixLength(
                  filterChain.filterChainMatch(), address, forDestination);

          if (currentMatchingPrefixLen >= 0) {
            if (currentMatchingPrefixLen < topMatchingPrefixLen) {
              continue;
            }
            if (currentMatchingPrefixLen > topMatchingPrefixLen) {
              topMatchingPrefixLen = currentMatchingPrefixLen;
              topOnes.clear();
            }
            topOnes.add(filterChain);
          }
        }
        return topOnes;
      }

      @Override
      public String toString() {
        return MoreObjects.toStringHelper(this)
            .add("routingConfigs", routingConfigs)
            .add("defaultSslContextProviderSupplier", defaultSslContextProviderSupplier)
            .add("defaultRoutingConfig", defaultRoutingConfig)
            .toString();
      }
    }
  }

  static final class FilterChainMatchingNegotiatorServerFactory
          implements InternalProtocolNegotiator.ServerFactory {
    private final InternalProtocolNegotiator.ServerFactory delegate;

    public FilterChainMatchingNegotiatorServerFactory(
            InternalProtocolNegotiator.ServerFactory delegate) {
      this.delegate = checkNotNull(delegate, "delegate");
    }

    @Override
    public ProtocolNegotiator newNegotiator(
            final ObjectPool<? extends Executor> offloadExecutorPool) {

      class FilterChainMatchingNegotiator implements ProtocolNegotiator {

        @Override
        public AsciiString scheme() {
          return SCHEME;
        }

        @Override
        public ChannelHandler newHandler(GrpcHttp2ConnectionHandler grpcHandler) {
          FilterChainSelectorManager filterChainSelectorManager =
              grpcHandler.getEagAttributes().get(ATTR_FILTER_CHAIN_SELECTOR_MANAGER);
          checkNotNull(filterChainSelectorManager, "filterChainSelectorManager");
          return new FilterChainMatchingHandler(grpcHandler, filterChainSelectorManager,
                  delegate.newNegotiator(offloadExecutorPool));
        }

        @Override
        public void close() {
        }
      }

      return new FilterChainMatchingNegotiator();
    }
  }

  /**
   * The FilterChain level configuration.
   */
  private static final class SelectedConfig {
    private final AtomicReference<ServerRoutingConfig> routingConfig;
    @Nullable
    private final SslContextProviderSupplier sslContextProviderSupplier;

    private SelectedConfig(AtomicReference<ServerRoutingConfig> routingConfig,
                           @Nullable SslContextProviderSupplier sslContextProviderSupplier) {
      this.routingConfig = checkNotNull(routingConfig, "routingConfig");
      this.sslContextProviderSupplier = sslContextProviderSupplier;
    }
  }

  private static class FilterChainSelectorManagerDeregister implements ChannelFutureListener {
    private final FilterChainSelectorManager filterChainSelectorManager;
    private final FilterChainSelectorManager.Closer closer;

    public FilterChainSelectorManagerDeregister(
        FilterChainSelectorManager filterChainSelectorManager,
        FilterChainSelectorManager.Closer closer) {
      this.filterChainSelectorManager =
          checkNotNull(filterChainSelectorManager, "filterChainSelectorManager");
      this.closer = checkNotNull(closer, "closer");
    }

    @Override public void operationComplete(ChannelFuture future) throws Exception {
      filterChainSelectorManager.deregister(closer);
    }
  }

  private static class GracefullyShutdownChannelRunnable implements Runnable {
    private final Channel channel;
    private final long drainGraceTime;
    @Nullable
    private final TimeUnit drainGraceTimeUnit;

    public GracefullyShutdownChannelRunnable(
        Channel channel, long drainGraceTime, @Nullable TimeUnit drainGraceTimeUnit) {
      this.channel = checkNotNull(channel, "channel");
      this.drainGraceTime = drainGraceTime;
      this.drainGraceTimeUnit = drainGraceTimeUnit;
    }

    @Override public void run() {
      Object gracefulCloseCommand = InternalGracefulServerCloseCommand.create(
          "xds_drain", drainGraceTime, drainGraceTimeUnit);
      channel.writeAndFlush(gracefulCloseCommand)
          .addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
    }
  }
}

1	/*
2	* Copyright 2021 The gRPC Authors
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package io.grpc.xds;
18
19	import static com.google.common.base.Preconditions.checkNotNull;
20	import static io.grpc.xds.XdsAttributes.ATTR_DRAIN_GRACE_NANOS;
21	import static io.grpc.xds.XdsAttributes.ATTR_FILTER_CHAIN_SELECTOR_MANAGER;
22	import static io.grpc.xds.XdsServerWrapper.ATTR_SERVER_ROUTING_CONFIG;
23	import static io.grpc.xds.internal.security.SecurityProtocolNegotiators.ATTR_SERVER_SSL_CONTEXT_PROVIDER_SUPPLIER;
24
25	import com.google.common.annotations.VisibleForTesting;
26	import com.google.common.base.MoreObjects;
27	import com.google.common.base.Strings;
28	import com.google.common.collect.Iterables;
29	import com.google.protobuf.UInt32Value;
30	import io.grpc.Attributes;
31	import io.grpc.internal.ObjectPool;
32	import io.grpc.netty.GrpcHttp2ConnectionHandler;
33	import io.grpc.netty.InternalGracefulServerCloseCommand;
34	import io.grpc.netty.InternalProtocolNegotiationEvent;
35	import io.grpc.netty.InternalProtocolNegotiator;
36	import io.grpc.netty.InternalProtocolNegotiator.ProtocolNegotiator;
37	import io.grpc.netty.ProtocolNegotiationEvent;
38	import io.grpc.xds.EnvoyServerProtoData.CidrRange;
39	import io.grpc.xds.EnvoyServerProtoData.ConnectionSourceType;
40	import io.grpc.xds.EnvoyServerProtoData.FilterChain;
41	import io.grpc.xds.EnvoyServerProtoData.FilterChainMatch;
42	import io.grpc.xds.XdsServerWrapper.ServerRoutingConfig;
43	import io.grpc.xds.internal.Matchers.CidrMatcher;
44	import io.grpc.xds.internal.security.SslContextProviderSupplier;
45	import io.netty.channel.Channel;
46	import io.netty.channel.ChannelFuture;
47	import io.netty.channel.ChannelFutureListener;
48	import io.netty.channel.ChannelHandler;
49	import io.netty.channel.ChannelHandlerContext;
50	import io.netty.channel.ChannelInboundHandlerAdapter;
51	import io.netty.util.AsciiString;
52	import java.net.Inet6Address;
53	import java.net.InetAddress;
54	import java.net.InetSocketAddress;
55	import java.util.ArrayList;
56	import java.util.Collection;
57	import java.util.Collections;
58	import java.util.List;
59	import java.util.Map;
60	import java.util.concurrent.Executor;
61	import java.util.concurrent.TimeUnit;
62	import java.util.concurrent.atomic.AtomicReference;
63	import java.util.logging.Level;
64	import java.util.logging.Logger;
65	import javax.annotation.Nullable;
66
67
68	/**
69	* Handles L4 filter chain match for the connection based on the xds configuration.
70	* */
71	@SuppressWarnings("FutureReturnValueIgnored") // Netty doesn't follow this pattern
72	final class FilterChainMatchingProtocolNegotiators {
73	private static final Logger log = Logger.getLogger(	1✔
74	FilterChainMatchingProtocolNegotiators.class.getName());	1✔
75
76	private static final AsciiString SCHEME = AsciiString.of("http");	1✔
77
78	private FilterChainMatchingProtocolNegotiators() {
79	}
80
81	@VisibleForTesting
82	static final class FilterChainMatchingHandler extends ChannelInboundHandlerAdapter {
83
84	private final GrpcHttp2ConnectionHandler grpcHandler;
85	private final FilterChainSelectorManager filterChainSelectorManager;
86	private final ProtocolNegotiator delegate;
87
88	FilterChainMatchingHandler(
89	GrpcHttp2ConnectionHandler grpcHandler,
90	FilterChainSelectorManager filterChainSelectorManager,
91	ProtocolNegotiator delegate) {	1✔
92	this.grpcHandler = checkNotNull(grpcHandler, "grpcHandler");	1✔
93	this.filterChainSelectorManager =	1✔
94	checkNotNull(filterChainSelectorManager, "filterChainSelectorManager");	1✔
95	this.delegate = checkNotNull(delegate, "delegate");	1✔
96	}	1✔
97
98	@Override
99	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
100	if (!(evt instanceof ProtocolNegotiationEvent)) {	1✔
101	super.userEventTriggered(ctx, evt);	×
102	return;	×
103	}
104	long drainGraceTime = 0;	1✔
105	TimeUnit drainGraceTimeUnit = null;	1✔
106	Long drainGraceNanosObj = grpcHandler.getEagAttributes().get(ATTR_DRAIN_GRACE_NANOS);	1✔
107	if (drainGraceNanosObj != null) {	1✔
108	drainGraceTime = drainGraceNanosObj;	1✔
109	drainGraceTimeUnit = TimeUnit.NANOSECONDS;	1✔
110	}
111	FilterChainSelectorManager.Closer closer = new FilterChainSelectorManager.Closer(	1✔
112	new GracefullyShutdownChannelRunnable(ctx.channel(), drainGraceTime, drainGraceTimeUnit));	1✔
113	FilterChainSelector selector = filterChainSelectorManager.register(closer);	1✔
114	ctx.channel().closeFuture().addListener(	1✔
115	new FilterChainSelectorManagerDeregister(filterChainSelectorManager, closer));
116	checkNotNull(selector, "selector");	1✔
117	SelectedConfig config = selector.select(	1✔
118	(InetSocketAddress) ctx.channel().localAddress(),	1✔
119	(InetSocketAddress) ctx.channel().remoteAddress());	1✔
120	if (config == null) {	1✔
121	log.log(Level.WARNING, "Connection from {0} to {1} has no matching filter chain. Closing",	1✔
122	new Object[] {ctx.channel().remoteAddress(), ctx.channel().localAddress()});	1✔
123	ctx.close().addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);	1✔
124	return;	1✔
125	}
126	ProtocolNegotiationEvent pne = (ProtocolNegotiationEvent) evt;	1✔
127	// TODO(zivy): merge into one key and take care of this outer class visibility.
128	Attributes attr = InternalProtocolNegotiationEvent.getAttributes(pne).toBuilder()	1✔
129	.set(ATTR_SERVER_SSL_CONTEXT_PROVIDER_SUPPLIER, config.sslContextProviderSupplier)	1✔
130	.set(ATTR_SERVER_ROUTING_CONFIG, config.routingConfig)	1✔
131	.build();	1✔
132	pne = InternalProtocolNegotiationEvent.withAttributes(pne, attr);	1✔
133	ctx.pipeline().replace(this, null, delegate.newHandler(grpcHandler));	1✔
134	ctx.fireUserEventTriggered(pne);	1✔
135	}	1✔
136
137	static final class FilterChainSelector {
138	public static final FilterChainSelector NO_FILTER_CHAIN = new FilterChainSelector(	1✔
139	Collections.<FilterChain, AtomicReference<ServerRoutingConfig>>emptyMap(),	1✔
140	null, new AtomicReference<ServerRoutingConfig>());
141	private final Map<FilterChain, AtomicReference<ServerRoutingConfig>> routingConfigs;
142	@Nullable
143	private final SslContextProviderSupplier defaultSslContextProviderSupplier;
144	private final AtomicReference<ServerRoutingConfig> defaultRoutingConfig;
145
146	FilterChainSelector(Map<FilterChain, AtomicReference<ServerRoutingConfig>> routingConfigs,
147	@Nullable SslContextProviderSupplier defaultSslContextProviderSupplier,
148	AtomicReference<ServerRoutingConfig> defaultRoutingConfig) {	1✔
149	this.routingConfigs = checkNotNull(routingConfigs, "routingConfigs");	1✔
150	this.defaultSslContextProviderSupplier = defaultSslContextProviderSupplier;	1✔
151	this.defaultRoutingConfig = checkNotNull(defaultRoutingConfig, "defaultRoutingConfig");	1✔
152	}	1✔
153
154	FilterChainSelector(Map<FilterChain, AtomicReference<ServerRoutingConfig>> routingConfigs) {
155	this(routingConfigs, null, new AtomicReference<>());	1✔
156	}	1✔
157
158	@VisibleForTesting
159	Map<FilterChain, AtomicReference<ServerRoutingConfig>> getRoutingConfigs() {
160	return routingConfigs;	1✔
161	}
162
163	@VisibleForTesting
164	AtomicReference<ServerRoutingConfig> getDefaultRoutingConfig() {
165	return defaultRoutingConfig;	1✔
166	}
167
168	@VisibleForTesting
169	SslContextProviderSupplier getDefaultSslContextProviderSupplier() {
170	return defaultSslContextProviderSupplier;	1✔
171	}
172
173	/**
174	* Throws IllegalStateException when no exact one match, and we should close the connection.
175	*/
176	SelectedConfig select(InetSocketAddress localAddr, InetSocketAddress remoteAddr) {
177	Collection<FilterChain> filterChains = routingConfigs.keySet();	1✔
178	filterChains = filterOnDestinationPort(filterChains);	1✔
179	filterChains = filterOnIpAddress(filterChains, localAddr.getAddress(), true);	1✔
180	filterChains = filterOnServerNames(filterChains);	1✔
181	filterChains = filterOnTransportProtocol(filterChains);	1✔
182	filterChains = filterOnApplicationProtocols(filterChains);	1✔
183	filterChains =	1✔
184	filterOnSourceType(filterChains, remoteAddr.getAddress(), localAddr.getAddress());	1✔
185	filterChains = filterOnIpAddress(filterChains, remoteAddr.getAddress(), false);	1✔
186	filterChains = filterOnSourcePort(filterChains, remoteAddr.getPort());	1✔
187
188	if (filterChains.size() > 1) {	1✔
189	throw new IllegalStateException("Found more than one matching filter chains. This should "	1✔
190	+ "not be possible as ClientXdsClient validated the chains for uniqueness.");
191	}
192	if (filterChains.size() == 1) {	1✔
193	FilterChain selected = Iterables.getOnlyElement(filterChains);	1✔
194	return new SelectedConfig(	1✔
195	routingConfigs.get(selected), selected.sslContextProviderSupplier());	1✔
196	}
197	if (defaultRoutingConfig.get() != null) {	1✔
198	return new SelectedConfig(defaultRoutingConfig, defaultSslContextProviderSupplier);	1✔
199	}
200	return null;	1✔
201	}
202
203	// reject if filer-chain-match has non-empty application_protocols
204	private static Collection<FilterChain> filterOnApplicationProtocols(
205	Collection<FilterChain> filterChains) {
206	ArrayList<FilterChain> filtered = new ArrayList<>(filterChains.size());	1✔
207	for (FilterChain filterChain : filterChains) {	1✔
208	FilterChainMatch filterChainMatch = filterChain.filterChainMatch();	1✔
209
210	if (filterChainMatch.applicationProtocols().isEmpty()) {	1✔
211	filtered.add(filterChain);	1✔
212	}
213	}	1✔
214	return filtered;	1✔
215	}
216
217	// reject if filer-chain-match has non-empty transport protocol other than "raw_buffer"
218	private static Collection<FilterChain> filterOnTransportProtocol(
219	Collection<FilterChain> filterChains) {
220	ArrayList<FilterChain> filtered = new ArrayList<>(filterChains.size());	1✔
221	for (FilterChain filterChain : filterChains) {	1✔
222	FilterChainMatch filterChainMatch = filterChain.filterChainMatch();	1✔
223
224	String transportProtocol = filterChainMatch.transportProtocol();	1✔
225	if (Strings.isNullOrEmpty(transportProtocol) \|\| "raw_buffer".equals(transportProtocol)) {	1✔
226	filtered.add(filterChain);	1✔
227	}
228	}	1✔
229	return filtered;	1✔
230	}
231
232	// reject if filer-chain-match has server_name(s)
233	private static Collection<FilterChain> filterOnServerNames(
234	Collection<FilterChain> filterChains) {
235	ArrayList<FilterChain> filtered = new ArrayList<>(filterChains.size());	1✔
236	for (FilterChain filterChain : filterChains) {	1✔
237	FilterChainMatch filterChainMatch = filterChain.filterChainMatch();	1✔
238
239	if (filterChainMatch.serverNames().isEmpty()) {	1✔
240	filtered.add(filterChain);	1✔
241	}
242	}	1✔
243	return filtered;	1✔
244	}
245
246	// destination_port present => Always fail match
247	private static Collection<FilterChain> filterOnDestinationPort(
248	Collection<FilterChain> filterChains) {
249	ArrayList<FilterChain> filtered = new ArrayList<>(filterChains.size());	1✔
250	for (FilterChain filterChain : filterChains) {	1✔
251	FilterChainMatch filterChainMatch = filterChain.filterChainMatch();	1✔
252
253	if (filterChainMatch.destinationPort()	1✔
254	== UInt32Value.getDefaultInstance().getValue()) {	1✔
255	filtered.add(filterChain);	1✔
256	}
257	}	1✔
258	return filtered;	1✔
259	}
260
261	private static Collection<FilterChain> filterOnSourcePort(
262	Collection<FilterChain> filterChains, int sourcePort) {
263	ArrayList<FilterChain> filteredOnMatch = new ArrayList<>(filterChains.size());	1✔
264	ArrayList<FilterChain> filteredOnEmpty = new ArrayList<>(filterChains.size());	1✔
265	for (FilterChain filterChain : filterChains) {	1✔
266	FilterChainMatch filterChainMatch = filterChain.filterChainMatch();	1✔
267
268	List<Integer> sourcePortsToMatch = filterChainMatch.sourcePorts();	1✔
269	if (sourcePortsToMatch.isEmpty()) {	1✔
270	filteredOnEmpty.add(filterChain);	1✔
271	} else if (sourcePortsToMatch.contains(sourcePort)) {	1✔
272	filteredOnMatch.add(filterChain);	1✔
273	}
274	}	1✔
275	// match against source port is more specific than match against empty list
276	return filteredOnMatch.isEmpty() ? filteredOnEmpty : filteredOnMatch;	1✔
277	}
278
279	private static Collection<FilterChain> filterOnSourceType(
280	Collection<FilterChain> filterChains, InetAddress sourceAddress,
281	InetAddress destAddress) {
282	ArrayList<FilterChain> filtered = new ArrayList<>(filterChains.size());	1✔
283	for (FilterChain filterChain : filterChains) {	1✔
284	FilterChainMatch filterChainMatch = filterChain.filterChainMatch();	1✔
285	ConnectionSourceType sourceType =	1✔
286	filterChainMatch.connectionSourceType();	1✔
287
288	boolean matching = false;	1✔
289	if (sourceType == ConnectionSourceType.SAME_IP_OR_LOOPBACK) {	1✔
290	matching =	1✔
291	sourceAddress.isLoopbackAddress()	1✔
292	\|\| sourceAddress.isAnyLocalAddress()	1✔
293	\|\| sourceAddress.equals(destAddress);	1✔
294	} else if (sourceType == ConnectionSourceType.EXTERNAL) {	1✔
295	matching = !sourceAddress.isLoopbackAddress() && !sourceAddress.isAnyLocalAddress();	1✔
296	} else { // ANY or null
297	matching = true;	1✔
298	}
299	if (matching) {	1✔
300	filtered.add(filterChain);	1✔
301	}
302	}	1✔
303	return filtered;	1✔
304	}
305
306	private static int getMatchingPrefixLength(
307	FilterChainMatch filterChainMatch, InetAddress address, boolean forDestination) {
308	boolean isIPv6 = address instanceof Inet6Address;	1✔
309	List<CidrRange> cidrRanges =
310	forDestination	1✔
311	? filterChainMatch.prefixRanges()	1✔
312	: filterChainMatch.sourcePrefixRanges();	1✔
313	int matchingPrefixLength;
314	if (cidrRanges.isEmpty()) { // if there is no CidrRange assume 0-length match	1✔
315	matchingPrefixLength = 0;	1✔
316	} else {
317	matchingPrefixLength = -1;	1✔
318	for (CidrRange cidrRange : cidrRanges) {	1✔
319	InetAddress cidrAddr = cidrRange.addressPrefix();	1✔
320	boolean cidrIsIpv6 = cidrAddr instanceof Inet6Address;	1✔
321	if (isIPv6 == cidrIsIpv6) {	1✔
322	int prefixLen = cidrRange.prefixLen();	1✔
323	CidrMatcher matcher = CidrMatcher.create(cidrAddr, prefixLen);	1✔
324	if (matcher.matches(address) && prefixLen > matchingPrefixLength) {	1✔
325	matchingPrefixLength = prefixLen;	1✔
326	}
327	}
328	}	1✔
329	}
330	return matchingPrefixLength;	1✔
331	}
332
333	// use prefix_ranges (CIDR) and get the most specific matches
334	private static Collection<FilterChain> filterOnIpAddress(
335	Collection<FilterChain> filterChains, InetAddress address, boolean forDestination) {
336	// current list of top ones
337	ArrayList<FilterChain> topOnes = new ArrayList<>(filterChains.size());	1✔
338	int topMatchingPrefixLen = -1;	1✔
339	for (FilterChain filterChain : filterChains) {	1✔
340	int currentMatchingPrefixLen = getMatchingPrefixLength(	1✔
341	filterChain.filterChainMatch(), address, forDestination);	1✔
342
343	if (currentMatchingPrefixLen >= 0) {	1✔
344	if (currentMatchingPrefixLen < topMatchingPrefixLen) {	1✔
345	continue;	1✔
346	}
347	if (currentMatchingPrefixLen > topMatchingPrefixLen) {	1✔
348	topMatchingPrefixLen = currentMatchingPrefixLen;	1✔
349	topOnes.clear();	1✔
350	}
351	topOnes.add(filterChain);	1✔
352	}
353	}	1✔
354	return topOnes;	1✔
355	}
356
357	@Override
358	public String toString() {
359	return MoreObjects.toStringHelper(this)	×
360	.add("routingConfigs", routingConfigs)	×
361	.add("defaultSslContextProviderSupplier", defaultSslContextProviderSupplier)	×
362	.add("defaultRoutingConfig", defaultRoutingConfig)	×
363	.toString();	×
364	}
365	}
366	}
367
368	static final class FilterChainMatchingNegotiatorServerFactory
369	implements InternalProtocolNegotiator.ServerFactory {
370	private final InternalProtocolNegotiator.ServerFactory delegate;
371
372	public FilterChainMatchingNegotiatorServerFactory(
373	InternalProtocolNegotiator.ServerFactory delegate) {	1✔
374	this.delegate = checkNotNull(delegate, "delegate");	1✔
375	}	1✔
376
377	@Override
378	public ProtocolNegotiator newNegotiator(
379	final ObjectPool<? extends Executor> offloadExecutorPool) {
380
381	class FilterChainMatchingNegotiator implements ProtocolNegotiator {	1✔
382
383	@Override
384	public AsciiString scheme() {
385	return SCHEME;	×
386	}
387
388	@Override
389	public ChannelHandler newHandler(GrpcHttp2ConnectionHandler grpcHandler) {
390	FilterChainSelectorManager filterChainSelectorManager =	1✔
391	grpcHandler.getEagAttributes().get(ATTR_FILTER_CHAIN_SELECTOR_MANAGER);	1✔
392	checkNotNull(filterChainSelectorManager, "filterChainSelectorManager");	1✔
393	return new FilterChainMatchingHandler(grpcHandler, filterChainSelectorManager,	1✔
394	delegate.newNegotiator(offloadExecutorPool));	1✔
395	}
396
397	@Override
398	public void close() {
399	}	1✔
400	}
401
402	return new FilterChainMatchingNegotiator();	1✔
403	}
404	}
405
406	/**
407	* The FilterChain level configuration.
408	*/
409	private static final class SelectedConfig {
410	private final AtomicReference<ServerRoutingConfig> routingConfig;
411	@Nullable
412	private final SslContextProviderSupplier sslContextProviderSupplier;
413
414	private SelectedConfig(AtomicReference<ServerRoutingConfig> routingConfig,
415	@Nullable SslContextProviderSupplier sslContextProviderSupplier) {	1✔
416	this.routingConfig = checkNotNull(routingConfig, "routingConfig");	1✔
417	this.sslContextProviderSupplier = sslContextProviderSupplier;	1✔
418	}	1✔
419	}
420
421	private static class FilterChainSelectorManagerDeregister implements ChannelFutureListener {
422	private final FilterChainSelectorManager filterChainSelectorManager;
423	private final FilterChainSelectorManager.Closer closer;
424
425	public FilterChainSelectorManagerDeregister(
426	FilterChainSelectorManager filterChainSelectorManager,
427	FilterChainSelectorManager.Closer closer) {	1✔
428	this.filterChainSelectorManager =	1✔
429	checkNotNull(filterChainSelectorManager, "filterChainSelectorManager");	1✔
430	this.closer = checkNotNull(closer, "closer");	1✔
431	}	1✔
432
433	@Override public void operationComplete(ChannelFuture future) throws Exception {
434	filterChainSelectorManager.deregister(closer);	1✔
435	}	1✔
436	}
437
438	private static class GracefullyShutdownChannelRunnable implements Runnable {
439	private final Channel channel;
440	private final long drainGraceTime;
441	@Nullable
442	private final TimeUnit drainGraceTimeUnit;
443
444	public GracefullyShutdownChannelRunnable(
445	Channel channel, long drainGraceTime, @Nullable TimeUnit drainGraceTimeUnit) {	1✔
446	this.channel = checkNotNull(channel, "channel");	1✔
447	this.drainGraceTime = drainGraceTime;	1✔
448	this.drainGraceTimeUnit = drainGraceTimeUnit;	1✔
449	}	1✔
450
451	@Override public void run() {
452	Object gracefulCloseCommand = InternalGracefulServerCloseCommand.create(	1✔
453	"xds_drain", drainGraceTime, drainGraceTimeUnit);
454	channel.writeAndFlush(gracefulCloseCommand)	1✔
455	.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);	1✔
456	}	1✔
457	}
458	}

grpc / grpc-java / #19717

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous