#18595

pending completion

Build # #18595

Build Type

push

github-actions

Committed by

ejona86

Commit Message

xds: handle the handlerRemoved callback to skip updateSslContext processing (#10118)

* xds: handle the handlerRemoved callback to skip updateSslContext processing
     In handlerAdded we submit a callback to updateSslContext but before the
     callback is executed the handler could be removed (e.g. bad connection)
     in which case the callback should skip all of the processing.
     Also added a unit test to check there is no exception.

Run Details

30835 of 34987 relevant lines covered (88.13%)

0.88 hits per line

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

89.06

/../xds/src/main/java/io/grpc/xds/internal/security/SecurityProtocolNegotiators.java

/*
 * Copyright 2019 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.internal.security;

import static com.google.common.base.Preconditions.checkNotNull;

import com.google.common.annotations.VisibleForTesting;
import io.grpc.Attributes;
import io.grpc.internal.GrpcUtil;
import io.grpc.internal.ObjectPool;
import io.grpc.netty.GrpcHttp2ConnectionHandler;
import io.grpc.netty.InternalProtocolNegotiationEvent;
import io.grpc.netty.InternalProtocolNegotiator;
import io.grpc.netty.InternalProtocolNegotiator.ProtocolNegotiator;
import io.grpc.netty.InternalProtocolNegotiators;
import io.grpc.netty.ProtocolNegotiationEvent;
import io.grpc.xds.InternalXdsAttributes;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.ssl.SslContext;
import io.netty.util.AsciiString;
import java.security.cert.CertStoreException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.Nullable;

/**
 * Provides client and server side gRPC {@link ProtocolNegotiator}s that use SDS to provide the SSL
 * context.
 */
@VisibleForTesting
public final class SecurityProtocolNegotiators {

  // Prevent instantiation.
  private SecurityProtocolNegotiators() {
  }

  private static final Logger logger
      = Logger.getLogger(SecurityProtocolNegotiators.class.getName());

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

  public static final Attributes.Key<SslContextProviderSupplier>
          ATTR_SERVER_SSL_CONTEXT_PROVIDER_SUPPLIER =
          Attributes.Key.create("io.grpc.xds.internal.sds.server.sslContextProviderSupplier");

  /**
   * Returns a {@link InternalProtocolNegotiator.ClientFactory}.
   *
   * @param fallbackNegotiator protocol negotiator to use as fallback.
   */
  public static InternalProtocolNegotiator.ClientFactory clientProtocolNegotiatorFactory(
      @Nullable InternalProtocolNegotiator.ClientFactory fallbackNegotiator) {
    return new ClientFactory(fallbackNegotiator);
  }

  public static InternalProtocolNegotiator.ServerFactory serverProtocolNegotiatorFactory(
      @Nullable InternalProtocolNegotiator.ServerFactory fallbackNegotiator) {
    return new ServerFactory(fallbackNegotiator);
  }

  private static final class ServerFactory implements InternalProtocolNegotiator.ServerFactory {

    private final InternalProtocolNegotiator.ServerFactory fallbackProtocolNegotiator;

    private ServerFactory(InternalProtocolNegotiator.ServerFactory fallbackNegotiator) {
      this.fallbackProtocolNegotiator = fallbackNegotiator;
    }

    @Override
    public ProtocolNegotiator newNegotiator(ObjectPool<? extends Executor> offloadExecutorPool) {
      return new ServerSdsProtocolNegotiator(
          fallbackProtocolNegotiator.newNegotiator(offloadExecutorPool));
    }
  }

  private static final class ClientFactory implements InternalProtocolNegotiator.ClientFactory {

    private final InternalProtocolNegotiator.ClientFactory fallbackProtocolNegotiator;

    private ClientFactory(InternalProtocolNegotiator.ClientFactory fallbackNegotiator) {
      this.fallbackProtocolNegotiator = fallbackNegotiator;
    }

    @Override
    public ProtocolNegotiator newNegotiator() {
      return new ClientSdsProtocolNegotiator(fallbackProtocolNegotiator.newNegotiator());
    }

    @Override
    public int getDefaultPort() {
      return GrpcUtil.DEFAULT_PORT_SSL;
    }
  }

  @VisibleForTesting
  static final class ClientSdsProtocolNegotiator implements ProtocolNegotiator {

    @Nullable private final ProtocolNegotiator fallbackProtocolNegotiator;

    ClientSdsProtocolNegotiator(@Nullable ProtocolNegotiator fallbackProtocolNegotiator) {
      this.fallbackProtocolNegotiator = fallbackProtocolNegotiator;
    }

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

    @Override
    public ChannelHandler newHandler(GrpcHttp2ConnectionHandler grpcHandler) {
      // check if SslContextProviderSupplier was passed via attributes
      SslContextProviderSupplier localSslContextProviderSupplier =
          grpcHandler.getEagAttributes().get(
              InternalXdsAttributes.ATTR_SSL_CONTEXT_PROVIDER_SUPPLIER);
      if (localSslContextProviderSupplier == null) {
        checkNotNull(
            fallbackProtocolNegotiator, "No TLS config and no fallbackProtocolNegotiator!");
        return fallbackProtocolNegotiator.newHandler(grpcHandler);
      }
      return new ClientSdsHandler(grpcHandler, localSslContextProviderSupplier);
    }

    @Override
    public void close() {}
  }

  private static class BufferReadsHandler extends ChannelInboundHandlerAdapter {
    private final List<Object> reads = new ArrayList<>();
    private boolean readComplete;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
      reads.add(msg);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
      readComplete = true;
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
      for (Object msg : reads) {
        super.channelRead(ctx, msg);
      }
      if (readComplete) {
        super.channelReadComplete(ctx);
      }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
      logger.log(Level.SEVERE, "exceptionCaught", cause);
      ctx.fireExceptionCaught(cause);
    }
  }

  @VisibleForTesting
  static final class ClientSdsHandler
      extends InternalProtocolNegotiators.ProtocolNegotiationHandler {
    private final GrpcHttp2ConnectionHandler grpcHandler;
    private final SslContextProviderSupplier sslContextProviderSupplier;

    ClientSdsHandler(
        GrpcHttp2ConnectionHandler grpcHandler,
        SslContextProviderSupplier sslContextProviderSupplier) {
      super(
          // superclass (InternalProtocolNegotiators.ProtocolNegotiationHandler) expects 'next'
          // handler but we don't have a next handler _yet_. So we "disable" superclass's behavior
          // here and then manually add 'next' when we call fireProtocolNegotiationEvent()
          new ChannelHandlerAdapter() {
            @Override
            public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
              ctx.pipeline().remove(this);
            }
          }, grpcHandler.getNegotiationLogger());
      checkNotNull(grpcHandler, "grpcHandler");
      this.grpcHandler = grpcHandler;
      this.sslContextProviderSupplier = sslContextProviderSupplier;
    }

    @Override
    protected void handlerAdded0(final ChannelHandlerContext ctx) {
      final BufferReadsHandler bufferReads = new BufferReadsHandler();
      ctx.pipeline().addBefore(ctx.name(), null, bufferReads);

      sslContextProviderSupplier.updateSslContext(
          new SslContextProvider.Callback(ctx.executor()) {

            @Override
            public void updateSslContext(SslContext sslContext) {
              if (ctx.isRemoved()) {
                return;
              }
              logger.log(
                  Level.FINEST,
                  "ClientSdsHandler.updateSslContext authority={0}, ctx.name={1}",
                  new Object[]{grpcHandler.getAuthority(), ctx.name()});
              ChannelHandler handler =
                  InternalProtocolNegotiators.tls(sslContext).newHandler(grpcHandler);

              // Delegate rest of handshake to TLS handler
              ctx.pipeline().addAfter(ctx.name(), null, handler);
              fireProtocolNegotiationEvent(ctx);
              ctx.pipeline().remove(bufferReads);
            }

            @Override
            public void onException(Throwable throwable) {
              ctx.fireExceptionCaught(throwable);
            }
          }
      );
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
        throws Exception {
      logger.log(Level.SEVERE, "exceptionCaught", cause);
      ctx.fireExceptionCaught(cause);
    }
  }

  private static final class ServerSdsProtocolNegotiator implements ProtocolNegotiator {

    @Nullable private final ProtocolNegotiator fallbackProtocolNegotiator;

    /** Constructor. */
    @VisibleForTesting
    public ServerSdsProtocolNegotiator(@Nullable ProtocolNegotiator fallbackProtocolNegotiator) {
      this.fallbackProtocolNegotiator = fallbackProtocolNegotiator;
    }

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

    @Override
    public ChannelHandler newHandler(GrpcHttp2ConnectionHandler grpcHandler) {
      return new HandlerPickerHandler(grpcHandler, fallbackProtocolNegotiator);
    }

    @Override
    public void close() {}
  }

  @VisibleForTesting
  static final class HandlerPickerHandler
      extends ChannelInboundHandlerAdapter {
    private final GrpcHttp2ConnectionHandler grpcHandler;
    @Nullable private final ProtocolNegotiator fallbackProtocolNegotiator;

    HandlerPickerHandler(
        GrpcHttp2ConnectionHandler grpcHandler,
        @Nullable ProtocolNegotiator fallbackProtocolNegotiator) {
      this.grpcHandler = checkNotNull(grpcHandler, "grpcHandler");
      this.fallbackProtocolNegotiator = fallbackProtocolNegotiator;
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
      if (evt instanceof ProtocolNegotiationEvent) {
        ProtocolNegotiationEvent pne = (ProtocolNegotiationEvent)evt;
        SslContextProviderSupplier sslContextProviderSupplier = InternalProtocolNegotiationEvent
                .getAttributes(pne).get(ATTR_SERVER_SSL_CONTEXT_PROVIDER_SUPPLIER);
        if (sslContextProviderSupplier == null) {
          logger.log(Level.FINE, "No sslContextProviderSupplier found in filterChainMatch "
              + "for connection from {0} to {1}",
              new Object[]{ctx.channel().remoteAddress(), ctx.channel().localAddress()});
          if (fallbackProtocolNegotiator == null) {
            ctx.fireExceptionCaught(new CertStoreException("No certificate source found!"));
            return;
          }
          logger.log(Level.FINE, "Using fallback credentials for connection from {0} to {1}",
              new Object[]{ctx.channel().remoteAddress(), ctx.channel().localAddress()});
          ctx.pipeline()
              .replace(
                  this,
                  null,
                  fallbackProtocolNegotiator.newHandler(grpcHandler));
          ctx.fireUserEventTriggered(pne);
          return;
        } else {
          ctx.pipeline()
              .replace(
                  this,
                  null,
                  new ServerSdsHandler(
                      grpcHandler, sslContextProviderSupplier));
          ctx.fireUserEventTriggered(pne);
          return;
        }
      } else {
        super.userEventTriggered(ctx, evt);
      }
    }
  }

  @VisibleForTesting
  static final class ServerSdsHandler
          extends InternalProtocolNegotiators.ProtocolNegotiationHandler {
    private final GrpcHttp2ConnectionHandler grpcHandler;
    private final SslContextProviderSupplier sslContextProviderSupplier;

    ServerSdsHandler(
            GrpcHttp2ConnectionHandler grpcHandler,
            SslContextProviderSupplier sslContextProviderSupplier) {
      super(
          // superclass (InternalProtocolNegotiators.ProtocolNegotiationHandler) expects 'next'
          // handler but we don't have a next handler _yet_. So we "disable" superclass's behavior
          // here and then manually add 'next' when we call fireProtocolNegotiationEvent()
          new ChannelHandlerAdapter() {
            @Override
            public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
              ctx.pipeline().remove(this);
            }
          }, grpcHandler.getNegotiationLogger());
      checkNotNull(grpcHandler, "grpcHandler");
      this.grpcHandler = grpcHandler;
      this.sslContextProviderSupplier = sslContextProviderSupplier;
    }

    @Override
    protected void handlerAdded0(final ChannelHandlerContext ctx) {
      final BufferReadsHandler bufferReads = new BufferReadsHandler();
      ctx.pipeline().addBefore(ctx.name(), null, bufferReads);

      sslContextProviderSupplier.updateSslContext(
          new SslContextProvider.Callback(ctx.executor()) {

            @Override
            public void updateSslContext(SslContext sslContext) {
              ChannelHandler handler =
                  InternalProtocolNegotiators.serverTls(sslContext).newHandler(grpcHandler);

              // Delegate rest of handshake to TLS handler
              if (!ctx.isRemoved()) {
                ctx.pipeline().addAfter(ctx.name(), null, handler);
                fireProtocolNegotiationEvent(ctx);
                ctx.pipeline().remove(bufferReads);
              }
            }

            @Override
            public void onException(Throwable throwable) {
              ctx.fireExceptionCaught(throwable);
            }
          }
      );
    }
  }
}

1	/*
2	* Copyright 2019 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.internal.security;
18
19	import static com.google.common.base.Preconditions.checkNotNull;
20
21	import com.google.common.annotations.VisibleForTesting;
22	import io.grpc.Attributes;
23	import io.grpc.internal.GrpcUtil;
24	import io.grpc.internal.ObjectPool;
25	import io.grpc.netty.GrpcHttp2ConnectionHandler;
26	import io.grpc.netty.InternalProtocolNegotiationEvent;
27	import io.grpc.netty.InternalProtocolNegotiator;
28	import io.grpc.netty.InternalProtocolNegotiator.ProtocolNegotiator;
29	import io.grpc.netty.InternalProtocolNegotiators;
30	import io.grpc.netty.ProtocolNegotiationEvent;
31	import io.grpc.xds.InternalXdsAttributes;
32	import io.netty.channel.ChannelHandler;
33	import io.netty.channel.ChannelHandlerAdapter;
34	import io.netty.channel.ChannelHandlerContext;
35	import io.netty.channel.ChannelInboundHandlerAdapter;
36	import io.netty.handler.ssl.SslContext;
37	import io.netty.util.AsciiString;
38	import java.security.cert.CertStoreException;
39	import java.util.ArrayList;
40	import java.util.List;
41	import java.util.concurrent.Executor;
42	import java.util.logging.Level;
43	import java.util.logging.Logger;
44	import javax.annotation.Nullable;
45
46	/**
47	* Provides client and server side gRPC {@link ProtocolNegotiator}s that use SDS to provide the SSL
48	* context.
49	*/
50	@VisibleForTesting
51	public final class SecurityProtocolNegotiators {
52
53	// Prevent instantiation.
54	private SecurityProtocolNegotiators() {
55	}
56
57	private static final Logger logger	1✔
58	= Logger.getLogger(SecurityProtocolNegotiators.class.getName());	1✔
59
60	private static final AsciiString SCHEME = AsciiString.of("http");	1✔
61
62	public static final Attributes.Key<SslContextProviderSupplier>
63	ATTR_SERVER_SSL_CONTEXT_PROVIDER_SUPPLIER =	1✔
64	Attributes.Key.create("io.grpc.xds.internal.sds.server.sslContextProviderSupplier");	1✔
65
66	/**
67	* Returns a {@link InternalProtocolNegotiator.ClientFactory}.
68	*
69	* @param fallbackNegotiator protocol negotiator to use as fallback.
70	*/
71	public static InternalProtocolNegotiator.ClientFactory clientProtocolNegotiatorFactory(
72	@Nullable InternalProtocolNegotiator.ClientFactory fallbackNegotiator) {
73	return new ClientFactory(fallbackNegotiator);	1✔
74	}
75
76	public static InternalProtocolNegotiator.ServerFactory serverProtocolNegotiatorFactory(
77	@Nullable InternalProtocolNegotiator.ServerFactory fallbackNegotiator) {
78	return new ServerFactory(fallbackNegotiator);	1✔
79	}
80
81	private static final class ServerFactory implements InternalProtocolNegotiator.ServerFactory {
82
83	private final InternalProtocolNegotiator.ServerFactory fallbackProtocolNegotiator;
84
85	private ServerFactory(InternalProtocolNegotiator.ServerFactory fallbackNegotiator) {	1✔
86	this.fallbackProtocolNegotiator = fallbackNegotiator;	1✔
87	}	1✔
88
89	@Override
90	public ProtocolNegotiator newNegotiator(ObjectPool<? extends Executor> offloadExecutorPool) {
91	return new ServerSdsProtocolNegotiator(	1✔
92	fallbackProtocolNegotiator.newNegotiator(offloadExecutorPool));	1✔
93	}
94	}
95
96	private static final class ClientFactory implements InternalProtocolNegotiator.ClientFactory {
97
98	private final InternalProtocolNegotiator.ClientFactory fallbackProtocolNegotiator;
99
100	private ClientFactory(InternalProtocolNegotiator.ClientFactory fallbackNegotiator) {	1✔
101	this.fallbackProtocolNegotiator = fallbackNegotiator;	1✔
102	}	1✔
103
104	@Override
105	public ProtocolNegotiator newNegotiator() {
106	return new ClientSdsProtocolNegotiator(fallbackProtocolNegotiator.newNegotiator());	1✔
107	}
108
109	@Override
110	public int getDefaultPort() {
111	return GrpcUtil.DEFAULT_PORT_SSL;	1✔
112	}
113	}
114
115	@VisibleForTesting
116	static final class ClientSdsProtocolNegotiator implements ProtocolNegotiator {
117
118	@Nullable private final ProtocolNegotiator fallbackProtocolNegotiator;
119
120	ClientSdsProtocolNegotiator(@Nullable ProtocolNegotiator fallbackProtocolNegotiator) {	1✔
121	this.fallbackProtocolNegotiator = fallbackProtocolNegotiator;	1✔
122	}	1✔
123
124	@Override
125	public AsciiString scheme() {
126	return SCHEME;	1✔
127	}
128
129	@Override
130	public ChannelHandler newHandler(GrpcHttp2ConnectionHandler grpcHandler) {
131	// check if SslContextProviderSupplier was passed via attributes
132	SslContextProviderSupplier localSslContextProviderSupplier =	1✔
133	grpcHandler.getEagAttributes().get(	1✔
134	InternalXdsAttributes.ATTR_SSL_CONTEXT_PROVIDER_SUPPLIER);
135	if (localSslContextProviderSupplier == null) {	1✔
136	checkNotNull(	1✔
137	fallbackProtocolNegotiator, "No TLS config and no fallbackProtocolNegotiator!");
138	return fallbackProtocolNegotiator.newHandler(grpcHandler);	1✔
139	}
140	return new ClientSdsHandler(grpcHandler, localSslContextProviderSupplier);	1✔
141	}
142
143	@Override
144	public void close() {}	1✔
145	}
146
147	private static class BufferReadsHandler extends ChannelInboundHandlerAdapter {	1✔
148	private final List<Object> reads = new ArrayList<>();	1✔
149	private boolean readComplete;
150
151	@Override
152	public void channelRead(ChannelHandlerContext ctx, Object msg) {
153	reads.add(msg);	1✔
154	}	1✔
155
156	@Override
157	public void channelReadComplete(ChannelHandlerContext ctx) {
158	readComplete = true;	1✔
159	}	1✔
160
161	@Override
162	public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
163	for (Object msg : reads) {	1✔
164	super.channelRead(ctx, msg);	1✔
165	}	1✔
166	if (readComplete) {	1✔
167	super.channelReadComplete(ctx);	1✔
168	}
169	}	1✔
170
171	@Override
172	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
173	logger.log(Level.SEVERE, "exceptionCaught", cause);	×
174	ctx.fireExceptionCaught(cause);	×
175	}	×
176	}
177
178	@VisibleForTesting
179	static final class ClientSdsHandler
180	extends InternalProtocolNegotiators.ProtocolNegotiationHandler {
181	private final GrpcHttp2ConnectionHandler grpcHandler;
182	private final SslContextProviderSupplier sslContextProviderSupplier;
183
184	ClientSdsHandler(
185	GrpcHttp2ConnectionHandler grpcHandler,
186	SslContextProviderSupplier sslContextProviderSupplier) {
187	super(	1✔
188	// superclass (InternalProtocolNegotiators.ProtocolNegotiationHandler) expects 'next'
189	// handler but we don't have a next handler _yet_. So we "disable" superclass's behavior
190	// here and then manually add 'next' when we call fireProtocolNegotiationEvent()
191	new ChannelHandlerAdapter() {	1✔
192	@Override
193	public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
194	ctx.pipeline().remove(this);	1✔
195	}	1✔
196	}, grpcHandler.getNegotiationLogger());	1✔
197	checkNotNull(grpcHandler, "grpcHandler");	1✔
198	this.grpcHandler = grpcHandler;	1✔
199	this.sslContextProviderSupplier = sslContextProviderSupplier;	1✔
200	}	1✔
201
202	@Override
203	protected void handlerAdded0(final ChannelHandlerContext ctx) {
204	final BufferReadsHandler bufferReads = new BufferReadsHandler();	1✔
205	ctx.pipeline().addBefore(ctx.name(), null, bufferReads);	1✔
206
207	sslContextProviderSupplier.updateSslContext(	1✔
208	new SslContextProvider.Callback(ctx.executor()) {	1✔
209
210	@Override
211	public void updateSslContext(SslContext sslContext) {
212	if (ctx.isRemoved()) {	1✔
213	return;	1✔
214	}
215	logger.log(	1✔
216	Level.FINEST,
217	"ClientSdsHandler.updateSslContext authority={0}, ctx.name={1}",
218	new Object[]{grpcHandler.getAuthority(), ctx.name()});	1✔
219	ChannelHandler handler =	1✔
220	InternalProtocolNegotiators.tls(sslContext).newHandler(grpcHandler);	1✔
221
222	// Delegate rest of handshake to TLS handler
223	ctx.pipeline().addAfter(ctx.name(), null, handler);	1✔
224	fireProtocolNegotiationEvent(ctx);	1✔
225	ctx.pipeline().remove(bufferReads);	1✔
226	}	1✔
227
228	@Override
229	public void onException(Throwable throwable) {
230	ctx.fireExceptionCaught(throwable);	×
231	}	×
232	}
233	);
234	}	1✔
235
236	@Override
237	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
238	throws Exception {
239	logger.log(Level.SEVERE, "exceptionCaught", cause);	×
240	ctx.fireExceptionCaught(cause);	×
241	}	×
242	}
243
244	private static final class ServerSdsProtocolNegotiator implements ProtocolNegotiator {
245
246	@Nullable private final ProtocolNegotiator fallbackProtocolNegotiator;
247
248	/** Constructor. */
249	@VisibleForTesting
250	public ServerSdsProtocolNegotiator(@Nullable ProtocolNegotiator fallbackProtocolNegotiator) {	1✔
251	this.fallbackProtocolNegotiator = fallbackProtocolNegotiator;	1✔
252	}	1✔
253
254	@Override
255	public AsciiString scheme() {
256	return SCHEME;	×
257	}
258
259	@Override
260	public ChannelHandler newHandler(GrpcHttp2ConnectionHandler grpcHandler) {
261	return new HandlerPickerHandler(grpcHandler, fallbackProtocolNegotiator);	1✔
262	}
263
264	@Override
265	public void close() {}	×
266	}
267
268	@VisibleForTesting
269	static final class HandlerPickerHandler
270	extends ChannelInboundHandlerAdapter {
271	private final GrpcHttp2ConnectionHandler grpcHandler;
272	@Nullable private final ProtocolNegotiator fallbackProtocolNegotiator;
273
274	HandlerPickerHandler(
275	GrpcHttp2ConnectionHandler grpcHandler,
276	@Nullable ProtocolNegotiator fallbackProtocolNegotiator) {	1✔
277	this.grpcHandler = checkNotNull(grpcHandler, "grpcHandler");	1✔
278	this.fallbackProtocolNegotiator = fallbackProtocolNegotiator;	1✔
279	}	1✔
280
281	@Override
282	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
283	if (evt instanceof ProtocolNegotiationEvent) {	1✔
284	ProtocolNegotiationEvent pne = (ProtocolNegotiationEvent)evt;	1✔
285	SslContextProviderSupplier sslContextProviderSupplier = InternalProtocolNegotiationEvent	1✔
286	.getAttributes(pne).get(ATTR_SERVER_SSL_CONTEXT_PROVIDER_SUPPLIER);	1✔
287	if (sslContextProviderSupplier == null) {	1✔
288	logger.log(Level.FINE, "No sslContextProviderSupplier found in filterChainMatch "	1✔
289	+ "for connection from {0} to {1}",
290	new Object[]{ctx.channel().remoteAddress(), ctx.channel().localAddress()});	1✔
291	if (fallbackProtocolNegotiator == null) {	1✔
292	ctx.fireExceptionCaught(new CertStoreException("No certificate source found!"));	1✔
293	return;	1✔
294	}
295	logger.log(Level.FINE, "Using fallback credentials for connection from {0} to {1}",	1✔
296	new Object[]{ctx.channel().remoteAddress(), ctx.channel().localAddress()});	1✔
297	ctx.pipeline()	1✔
298	.replace(	1✔
299	this,
300	null,
301	fallbackProtocolNegotiator.newHandler(grpcHandler));	1✔
302	ctx.fireUserEventTriggered(pne);	1✔
303	return;	1✔
304	} else {
305	ctx.pipeline()	1✔
306	.replace(	1✔
307	this,
308	null,
309	new ServerSdsHandler(
310	grpcHandler, sslContextProviderSupplier));
311	ctx.fireUserEventTriggered(pne);	1✔
312	return;	1✔
313	}
314	} else {
315	super.userEventTriggered(ctx, evt);	×
316	}
317	}	×
318	}
319
320	@VisibleForTesting
321	static final class ServerSdsHandler
322	extends InternalProtocolNegotiators.ProtocolNegotiationHandler {
323	private final GrpcHttp2ConnectionHandler grpcHandler;
324	private final SslContextProviderSupplier sslContextProviderSupplier;
325
326	ServerSdsHandler(
327	GrpcHttp2ConnectionHandler grpcHandler,
328	SslContextProviderSupplier sslContextProviderSupplier) {
329	super(	1✔
330	// superclass (InternalProtocolNegotiators.ProtocolNegotiationHandler) expects 'next'
331	// handler but we don't have a next handler _yet_. So we "disable" superclass's behavior
332	// here and then manually add 'next' when we call fireProtocolNegotiationEvent()
333	new ChannelHandlerAdapter() {	1✔
334	@Override
335	public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
336	ctx.pipeline().remove(this);	1✔
337	}	1✔
338	}, grpcHandler.getNegotiationLogger());	1✔
339	checkNotNull(grpcHandler, "grpcHandler");	1✔
340	this.grpcHandler = grpcHandler;	1✔
341	this.sslContextProviderSupplier = sslContextProviderSupplier;	1✔
342	}	1✔
343
344	@Override
345	protected void handlerAdded0(final ChannelHandlerContext ctx) {
346	final BufferReadsHandler bufferReads = new BufferReadsHandler();	1✔
347	ctx.pipeline().addBefore(ctx.name(), null, bufferReads);	1✔
348
349	sslContextProviderSupplier.updateSslContext(	1✔
350	new SslContextProvider.Callback(ctx.executor()) {	1✔
351
352	@Override
353	public void updateSslContext(SslContext sslContext) {
354	ChannelHandler handler =	1✔
355	InternalProtocolNegotiators.serverTls(sslContext).newHandler(grpcHandler);	1✔
356
357	// Delegate rest of handshake to TLS handler
358	if (!ctx.isRemoved()) {	1✔
359	ctx.pipeline().addAfter(ctx.name(), null, handler);	1✔
360	fireProtocolNegotiationEvent(ctx);	1✔
361	ctx.pipeline().remove(bufferReads);	1✔
362	}
363	}	1✔
364
365	@Override
366	public void onException(Throwable throwable) {
367	ctx.fireExceptionCaught(throwable);	×
368	}	×
369	}
370	);
371	}	1✔
372	}
373	}

grpc / grpc-java / #18595

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