Skip to content

Commit 4d82b30

Browse files
dingshuangxi888徒钟
dingshuangxi888
and
徒钟
authored
[ISSUE apache#6866] Move the judgment logic of grpc TLS mode to improve the scalability of ProtocolNegotiator (apache#6867)
[ISSUE apache#6866] Move the judgment logic of grpc TLS mode to improve the scalability of ProtocolNegotiator --------- Co-authored-by: 徒钟 <[email protected]>
1 parent f10f905 commit 4d82b30

File tree

2 files changed

+73
-79
lines changed

2 files changed

+73
-79
lines changed

Diff for: ‎proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java

+1-58
Original file line numberDiff line numberDiff line change
@@ -19,21 +19,11 @@
1919
import io.grpc.BindableService;
2020
import io.grpc.ServerInterceptor;
2121
import io.grpc.ServerServiceDefinition;
22-
import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
2322
import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
2423
import io.grpc.netty.shaded.io.netty.channel.epoll.EpollEventLoopGroup;
2524
import io.grpc.netty.shaded.io.netty.channel.epoll.EpollServerSocketChannel;
2625
import io.grpc.netty.shaded.io.netty.channel.nio.NioEventLoopGroup;
2726
import io.grpc.netty.shaded.io.netty.channel.socket.nio.NioServerSocketChannel;
28-
import io.grpc.netty.shaded.io.netty.handler.ssl.ClientAuth;
29-
import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
30-
import io.grpc.netty.shaded.io.netty.handler.ssl.util.InsecureTrustManagerFactory;
31-
import io.grpc.netty.shaded.io.netty.handler.ssl.util.SelfSignedCertificate;
32-
import java.io.IOException;
33-
import java.io.InputStream;
34-
import java.nio.file.Files;
35-
import java.nio.file.Paths;
36-
import java.security.cert.CertificateException;
3727
import java.util.List;
3828
import java.util.concurrent.ThreadPoolExecutor;
3929
import java.util.concurrent.TimeUnit;
@@ -44,13 +34,10 @@
4434
import org.apache.rocketmq.logging.org.slf4j.Logger;
4535
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
4636
import org.apache.rocketmq.proxy.config.ConfigurationManager;
47-
import org.apache.rocketmq.proxy.config.ProxyConfig;
4837
import org.apache.rocketmq.proxy.grpc.interceptor.AuthenticationInterceptor;
4938
import org.apache.rocketmq.proxy.grpc.interceptor.ContextInterceptor;
5039
import org.apache.rocketmq.proxy.grpc.interceptor.GlobalExceptionInterceptor;
5140
import org.apache.rocketmq.proxy.grpc.interceptor.HeaderInterceptor;
52-
import org.apache.rocketmq.remoting.common.TlsMode;
53-
import org.apache.rocketmq.remoting.netty.TlsSystemConfig;
5441

5542
public class GrpcServerBuilder {
5643
private static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
@@ -63,12 +50,7 @@ public static GrpcServerBuilder newBuilder(ThreadPoolExecutor executor, int port
6350
protected GrpcServerBuilder(ThreadPoolExecutor executor, int port) {
6451
serverBuilder = NettyServerBuilder.forPort(port);
6552

66-
try {
67-
configSslContext(serverBuilder);
68-
} catch (Exception e) {
69-
log.error("grpc tls set failed. msg: {}, e:", e.getMessage(), e);
70-
throw new RuntimeException("grpc tls set failed: " + e.getMessage());
71-
}
53+
serverBuilder.protocolNegotiator(new OptionalSSLProtocolNegotiator());
7254

7355
// build server
7456
int bossLoopNum = ConfigurationManager.getProxyConfig().getGrpcBossLoopNum();
@@ -116,45 +98,6 @@ public GrpcServer build() {
11698
return new GrpcServer(this.serverBuilder.build());
11799
}
118100

119-
protected void configSslContext(NettyServerBuilder serverBuilder) throws IOException, CertificateException {
120-
if (null == serverBuilder) {
121-
return;
122-
}
123-
124-
TlsMode tlsMode = TlsSystemConfig.tlsMode;
125-
if (!TlsMode.DISABLED.equals(tlsMode)) {
126-
SslContext sslContext = loadSslContext();
127-
if (TlsMode.PERMISSIVE.equals(tlsMode)) {
128-
serverBuilder.protocolNegotiator(new OptionalSSLProtocolNegotiator(sslContext));
129-
} else {
130-
serverBuilder.sslContext(sslContext);
131-
}
132-
}
133-
}
134-
135-
protected SslContext loadSslContext() throws CertificateException, IOException {
136-
ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
137-
if (proxyConfig.isTlsTestModeEnable()) {
138-
SelfSignedCertificate selfSignedCertificate = new SelfSignedCertificate();
139-
return GrpcSslContexts.forServer(selfSignedCertificate.certificate(), selfSignedCertificate.privateKey())
140-
.trustManager(InsecureTrustManagerFactory.INSTANCE)
141-
.clientAuth(ClientAuth.NONE)
142-
.build();
143-
} else {
144-
String tlsKeyPath = ConfigurationManager.getProxyConfig().getTlsKeyPath();
145-
String tlsCertPath = ConfigurationManager.getProxyConfig().getTlsCertPath();
146-
try (InputStream serverKeyInputStream = Files.newInputStream(Paths.get(tlsKeyPath));
147-
InputStream serverCertificateStream = Files.newInputStream(Paths.get(tlsCertPath))) {
148-
SslContext res = GrpcSslContexts.forServer(serverCertificateStream, serverKeyInputStream)
149-
.trustManager(InsecureTrustManagerFactory.INSTANCE)
150-
.clientAuth(ClientAuth.NONE)
151-
.build();
152-
log.info("load TLS configured OK");
153-
return res;
154-
}
155-
}
156-
}
157-
158101
public GrpcServerBuilder configInterceptor() {
159102
// grpc interceptors, including acl, logging etc.
160103
List<AccessValidator> accessValidators = ServiceProvider.load(AccessValidator.class);

Diff for: ‎proxy/src/main/java/org/apache/rocketmq/proxy/grpc/OptionalSSLProtocolNegotiator.java

+72-21
Original file line numberDiff line numberDiff line change
@@ -17,31 +17,44 @@
1717
package org.apache.rocketmq.proxy.grpc;
1818

1919
import io.grpc.netty.shaded.io.grpc.netty.GrpcHttp2ConnectionHandler;
20+
import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
2021
import io.grpc.netty.shaded.io.grpc.netty.InternalProtocolNegotiationEvent;
2122
import io.grpc.netty.shaded.io.grpc.netty.InternalProtocolNegotiator;
2223
import io.grpc.netty.shaded.io.grpc.netty.InternalProtocolNegotiators;
2324
import io.grpc.netty.shaded.io.netty.buffer.ByteBuf;
2425
import io.grpc.netty.shaded.io.netty.channel.ChannelHandler;
2526
import io.grpc.netty.shaded.io.netty.channel.ChannelHandlerContext;
2627
import io.grpc.netty.shaded.io.netty.handler.codec.ByteToMessageDecoder;
28+
import io.grpc.netty.shaded.io.netty.handler.ssl.ClientAuth;
2729
import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
2830
import io.grpc.netty.shaded.io.netty.handler.ssl.SslHandler;
31+
import io.grpc.netty.shaded.io.netty.handler.ssl.util.InsecureTrustManagerFactory;
32+
import io.grpc.netty.shaded.io.netty.handler.ssl.util.SelfSignedCertificate;
2933
import io.grpc.netty.shaded.io.netty.util.AsciiString;
34+
import java.io.InputStream;
35+
import java.nio.file.Files;
36+
import java.nio.file.Paths;
3037
import java.util.List;
3138
import org.apache.rocketmq.common.constant.LoggerName;
3239
import org.apache.rocketmq.logging.org.slf4j.Logger;
3340
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
41+
import org.apache.rocketmq.proxy.config.ConfigurationManager;
42+
import org.apache.rocketmq.proxy.config.ProxyConfig;
43+
import org.apache.rocketmq.remoting.common.TlsMode;
44+
import org.apache.rocketmq.remoting.netty.TlsSystemConfig;
3445

3546
public class OptionalSSLProtocolNegotiator implements InternalProtocolNegotiator.ProtocolNegotiator {
36-
private static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
37-
private final SslContext sslContext;
47+
protected static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
48+
3849
/**
3950
* the length of the ssl record header (in bytes)
4051
*/
4152
private static final int SSL_RECORD_HEADER_LENGTH = 5;
4253

43-
public OptionalSSLProtocolNegotiator(SslContext sslContext) {
44-
this.sslContext = sslContext;
54+
private static SslContext sslContext;
55+
56+
public OptionalSSLProtocolNegotiator() {
57+
sslContext = loadSslContext();
4558
}
4659

4760
@Override
@@ -50,43 +63,81 @@ public AsciiString scheme() {
5063
}
5164

5265
@Override
53-
public ChannelHandler newHandler(GrpcHttp2ConnectionHandler grpcHttp2ConnectionHandler) {
54-
ChannelHandler plaintext =
55-
InternalProtocolNegotiators.serverPlaintext().newHandler(grpcHttp2ConnectionHandler);
56-
ChannelHandler ssl =
57-
InternalProtocolNegotiators.serverTls(sslContext).newHandler(grpcHttp2ConnectionHandler);
58-
return new PortUnificationServerHandler(ssl, plaintext);
66+
public ChannelHandler newHandler(GrpcHttp2ConnectionHandler grpcHandler) {
67+
return new PortUnificationServerHandler(grpcHandler);
5968
}
6069

6170
@Override
6271
public void close() {}
6372

73+
private static SslContext loadSslContext() {
74+
try {
75+
ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
76+
if (proxyConfig.isTlsTestModeEnable()) {
77+
SelfSignedCertificate selfSignedCertificate = new SelfSignedCertificate();
78+
return GrpcSslContexts.forServer(selfSignedCertificate.certificate(),
79+
selfSignedCertificate.privateKey())
80+
.trustManager(InsecureTrustManagerFactory.INSTANCE)
81+
.clientAuth(ClientAuth.NONE)
82+
.build();
83+
} else {
84+
String tlsKeyPath = ConfigurationManager.getProxyConfig().getTlsKeyPath();
85+
String tlsCertPath = ConfigurationManager.getProxyConfig().getTlsCertPath();
86+
try (InputStream serverKeyInputStream = Files.newInputStream(
87+
Paths.get(tlsKeyPath));
88+
InputStream serverCertificateStream = Files.newInputStream(
89+
Paths.get(tlsCertPath))) {
90+
SslContext res = GrpcSslContexts.forServer(serverCertificateStream,
91+
serverKeyInputStream)
92+
.trustManager(InsecureTrustManagerFactory.INSTANCE)
93+
.clientAuth(ClientAuth.NONE)
94+
.build();
95+
log.info("grpc load TLS configured OK");
96+
return res;
97+
}
98+
}
99+
} catch (Exception e) {
100+
log.error("grpc tls set failed. msg: {}, e:", e.getMessage(), e);
101+
throw new RuntimeException("grpc tls set failed: " + e.getMessage());
102+
}
103+
}
104+
64105
public static class PortUnificationServerHandler extends ByteToMessageDecoder {
106+
65107
private final ChannelHandler ssl;
66108
private final ChannelHandler plaintext;
67109

68-
public PortUnificationServerHandler(ChannelHandler ssl, ChannelHandler plaintext) {
69-
this.ssl = ssl;
70-
this.plaintext = plaintext;
110+
public PortUnificationServerHandler(GrpcHttp2ConnectionHandler grpcHandler) {
111+
this.ssl = InternalProtocolNegotiators.serverTls(sslContext)
112+
.newHandler(grpcHandler);
113+
this.plaintext = InternalProtocolNegotiators.serverPlaintext()
114+
.newHandler(grpcHandler);
71115
}
72116

73117
@Override
74118
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
75-
throws Exception {
119+
throws Exception {
76120
try {
77-
// in SslHandler.isEncrypted, it need at least 5 bytes to judge is encrypted or not
78-
if (in.readableBytes() < SSL_RECORD_HEADER_LENGTH) {
79-
return;
80-
}
81-
if (SslHandler.isEncrypted(in)) {
121+
TlsMode tlsMode = TlsSystemConfig.tlsMode;
122+
if (TlsMode.ENFORCING.equals(tlsMode)) {
82123
ctx.pipeline().addAfter(ctx.name(), null, this.ssl);
83-
} else {
124+
} else if (TlsMode.DISABLED.equals(tlsMode)) {
84125
ctx.pipeline().addAfter(ctx.name(), null, this.plaintext);
126+
} else {
127+
// in SslHandler.isEncrypted, it need at least 5 bytes to judge is encrypted or not
128+
if (in.readableBytes() < SSL_RECORD_HEADER_LENGTH) {
129+
return;
130+
}
131+
if (SslHandler.isEncrypted(in)) {
132+
ctx.pipeline().addAfter(ctx.name(), null, this.ssl);
133+
} else {
134+
ctx.pipeline().addAfter(ctx.name(), null, this.plaintext);
135+
}
85136
}
86137
ctx.fireUserEventTriggered(InternalProtocolNegotiationEvent.getDefault());
87138
ctx.pipeline().remove(this);
88139
} catch (Exception e) {
89-
log.error("process protocol negotiator failed.", e);
140+
log.error("process ssl protocol negotiator failed.", e);
90141
throw e;
91142
}
92143
}

0 commit comments

Comments
 (0)