zhuyijun/nacos
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ssl context reload spi (#10150)

Browse Source
This commit is contained in:
nov.lzf 2023-03-21 15:27:57 +08:00 committed by GitHub
parent 7df6f6d47c
commit c39ba4a35c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 410 additions and 183 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
common/src/main/java/com/alibaba/nacos/common/remote/client/grpc/GrpcClient.java
View File

@ -135,9 +135,10 @@ public abstract class GrpcClient extends RpcClient {
.setThreadPoolMaxSize(threadPoolMaxSize).setLabels(labels).build());
}
public GrpcClient(String name, Integer threadPoolCoreSize, Integer threadPoolMaxSize, Map<String, String> labels, RpcClientTlsConfig tlsConfig) {
this(DefaultGrpcClientConfig.newBuilder().setName(name).setThreadPoolCoreSize(threadPoolCoreSize).setTlsConfig(tlsConfig)
.setThreadPoolMaxSize(threadPoolMaxSize).setLabels(labels).build());
public GrpcClient(String name, Integer threadPoolCoreSize, Integer threadPoolMaxSize, Map<String, String> labels,
RpcClientTlsConfig tlsConfig) {
this(DefaultGrpcClientConfig.newBuilder().setName(name).setThreadPoolCoreSize(threadPoolCoreSize)
.setTlsConfig(tlsConfig).setThreadPoolMaxSize(threadPoolMaxSize).setLabels(labels).build());
}
protected ThreadPoolExecutor createGrpcExecutor(String serverIp) {
@ -181,8 +182,8 @@ public abstract class GrpcClient extends RpcClient {
private ManagedChannel createNewManagedChannel(String serverIp, int serverPort) {
LOGGER.info("grpc client connection server:{} ip,serverPort:{},grpcTslConfig:{}", serverIp, serverPort,
JacksonUtils.toJson(clientConfig.tlsConfig()));
ManagedChannelBuilder<?> managedChannelBuilder = buildChannel(serverIp, serverPort, buildSslContext())
.executor(grpcExecutor).compressorRegistry(CompressorRegistry.getDefaultInstance())
ManagedChannelBuilder<?> managedChannelBuilder = buildChannel(serverIp, serverPort, buildSslContext()).executor(
grpcExecutor).compressorRegistry(CompressorRegistry.getDefaultInstance())
.decompressorRegistry(DecompressorRegistry.getDefaultInstance())
.maxInboundMessageSize(clientConfig.maxInboundMessageSize())
.keepAliveTime(clientConfig.channelKeepAlive(), TimeUnit.MILLISECONDS)
@ -221,6 +222,11 @@ public abstract class GrpcClient extends RpcClient {
} catch (Exception e) {
LoggerUtils.printIfErrorEnabled(LOGGER,
"Server check fail, please check server {} ,port {} is available , error ={}", ip, port, e);
if (this.clientConfig != null && this.clientConfig.tlsConfig() != null && this.clientConfig.tlsConfig()
.getEnableTls()) {
LoggerUtils.printIfErrorEnabled(LOGGER,
"current client is require tls encrypted ,server must support tls ,please check");
}
return null;
}
}
@ -364,13 +370,11 @@ public abstract class GrpcClient extends RpcClient {
private ManagedChannelBuilder buildChannel(String serverIp, int port, Optional<SslContext> sslContext) {
if (sslContext.isPresent()) {
return NettyChannelBuilder.forAddress(serverIp, port)
.negotiationType(NegotiationType.TLS)
return NettyChannelBuilder.forAddress(serverIp, port).negotiationType(NegotiationType.TLS)
.sslContext(sslContext.get());
} else {
return ManagedChannelBuilder
.forAddress(serverIp, port).usePlaintext();
return ManagedChannelBuilder.forAddress(serverIp, port).usePlaintext();
}
}
@ -400,12 +404,14 @@ public abstract class GrpcClient extends RpcClient {
}
if (tlsConfig.getMutualAuthEnable()) {
if (StringUtils.isBlank(tlsConfig.getCertChainFile()) || StringUtils.isBlank(tlsConfig.getCertPrivateKey())) {
if (StringUtils.isBlank(tlsConfig.getCertChainFile()) || StringUtils.isBlank(
tlsConfig.getCertPrivateKey())) {
throw new IllegalArgumentException("client certChainFile or certPrivateKey must be not null");
}
Resource certChainFile = resourceLoader.getResource(tlsConfig.getCertChainFile());
Resource privateKey = resourceLoader.getResource(tlsConfig.getCertPrivateKey());
builder.keyManager(certChainFile.getInputStream(), privateKey.getInputStream(), tlsConfig.getCertPrivateKeyPassword());
builder.keyManager(certChainFile.getInputStream(), privateKey.getInputStream(),
tlsConfig.getCertPrivateKeyPassword());
}
return Optional.of(builder.build());
} catch (Exception e) {

33
core/src/main/java/com/alibaba/nacos/core/remote/BaseRpcServer.java
View File

@ -16,7 +16,6 @@
package com.alibaba.nacos.core.remote;
import com.alibaba.nacos.common.JustForTest;
import com.alibaba.nacos.common.remote.ConnectionType;
import com.alibaba.nacos.common.remote.PayloadRegistry;
import com.alibaba.nacos.common.utils.JacksonUtils;
@ -40,12 +39,7 @@ public abstract class BaseRpcServer {
}
@Autowired
protected RpcServerTlsConfig grpcServerConfig;
@JustForTest
public void setGrpcServerConfig(RpcServerTlsConfig grpcServerConfig) {
this.grpcServerConfig = grpcServerConfig;
}
protected RpcServerTlsConfig rpcServerTlsConfig;
/**
* Start sever.
@ -53,12 +47,18 @@ public abstract class BaseRpcServer {
@PostConstruct
public void start() throws Exception {
String serverName = getClass().getSimpleName();
String tlsConfig = JacksonUtils.toJson(grpcServerConfig);
Loggers.REMOTE.info("Nacos {} Rpc server starting at port {} and tls config:{}", serverName, getServicePort(), tlsConfig);
String tlsConfig = JacksonUtils.toJson(rpcServerTlsConfig);
Loggers.REMOTE.info("Nacos {} Rpc server starting at port {} and tls config:{}", serverName, getServicePort(),
tlsConfig);
startServer();
Loggers.REMOTE.info("Nacos {} Rpc server started at port {} and tls config:{}", serverName, getServicePort(), tlsConfig);
if (RpcServerSslContextRefresherHolder.getInstance() != null) {
RpcServerSslContextRefresherHolder.getInstance().refresh(this);
}
Loggers.REMOTE.info("Nacos {} Rpc server started at port {} and tls config:{}", serverName, getServicePort(),
tlsConfig);
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
Loggers.REMOTE.info("Nacos {} Rpc server stopping", serverName);
try {
@ -78,6 +78,19 @@ public abstract class BaseRpcServer {
*/
public abstract ConnectionType getConnectionType();
public RpcServerTlsConfig getRpcServerTlsConfig() {
return rpcServerTlsConfig;
}
public void setRpcServerTlsConfig(RpcServerTlsConfig rpcServerTlsConfig) {
this.rpcServerTlsConfig = rpcServerTlsConfig;
}
/**
* reload ssl context.
*/
public abstract void reloadSslContext();
/**
* Start sever.
*

41
core/src/main/java/com/alibaba/nacos/core/remote/RpcServerSslContextRefresher.java Normal file
View File

@ -0,0 +1,41 @@
package com.alibaba.nacos.core.remote;
/**
* ssl context refresher spi holder.
*
* @author liuzunfei
* @version $Id: RequestFilters.java, v 0.1 2023年03月17日 12:00 PM liuzunfei Exp $
*/
/*
* Copyright 1999-2020 Alibaba Group Holding Ltd.
*
* 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.
*/
public interface RpcServerSslContextRefresher {
/**
* listener current rpc server and do something on ssl context change.
*
* @param baseRpcServer rpc server.
* @return
*/
SslContextChangeAware refresh(BaseRpcServer baseRpcServer);
/**
* refresher name.
*
* @return
*/
String getName();
}

75
core/src/main/java/com/alibaba/nacos/core/remote/RpcServerSslContextRefresherHolder.java Normal file
View File

@ -0,0 +1,75 @@
/*
* Copyright 1999-2020 Alibaba Group Holding Ltd.
*
* 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 com.alibaba.nacos.core.remote;
import com.alibaba.nacos.common.spi.NacosServiceLoader;
import com.alibaba.nacos.common.utils.StringUtils;
import com.alibaba.nacos.core.utils.Loggers;
import com.alibaba.nacos.sys.utils.ApplicationUtils;
import java.util.Collection;
/**
* ssl context refresher spi holder.
*
* @author liuzunfei
* @version $Id: RequestFilters.java, v 0.1 2023年03月17日 12:00 PM liuzunfei Exp $
*/
public class RpcServerSslContextRefresherHolder {
private static RpcServerSslContextRefresher instance;
private static volatile boolean init = false;
public static RpcServerSslContextRefresher getInstance() {
if (init) {
return instance;
}
synchronized (RpcServerSslContextRefresherHolder.class) {
if (init) {
return instance;
}
RpcServerTlsConfig rpcServerTlsConfig = ApplicationUtils.getBean(RpcServerTlsConfig.class);
String sslContextRefresher = rpcServerTlsConfig.getSslContextRefresher();
if (StringUtils.isNotBlank(sslContextRefresher)) {
Collection<RpcServerSslContextRefresher> load = NacosServiceLoader.load(
RpcServerSslContextRefresher.class);
for (RpcServerSslContextRefresher contextRefresher : load) {
if (sslContextRefresher.equals(contextRefresher.getName())) {
instance = contextRefresher;
Loggers.REMOTE.info("RpcServerSslContextRefresher of Name {} Founded->{}", sslContextRefresher,
contextRefresher.getClass().getSimpleName());
break;
}
}
if (instance == null) {
Loggers.REMOTE.info("RpcServerSslContextRefresher of Name {} not found", sslContextRefresher);
}
} else {
Loggers.REMOTE.info(
"No RpcServerSslContextRefresher specified,Ssl Context auto refresh not supported.");
}
Loggers.REMOTE.info("RpcServerSslContextRefresher init end");
init = true;
}
return instance;
}
}

10
core/src/main/java/com/alibaba/nacos/core/remote/RpcServerTlsConfig.java
View File

@ -32,6 +32,8 @@ public class RpcServerTlsConfig extends TlsConfig {
public static final String PREFIX = "nacos.remote.server.rpc.tls";
private String sslContextRefresher = "";
private Boolean compatibility = true;
public Boolean getCompatibility() {
@ -41,4 +43,12 @@ public class RpcServerTlsConfig extends TlsConfig {
public void setCompatibility(Boolean compatibility) {
this.compatibility = compatibility;
}
public String getSslContextRefresher() {
return sslContextRefresher;
}
public void setSslContextRefresher(String sslContextRefresher) {
this.sslContextRefresher = sslContextRefresher;
}
}

43
core/src/main/java/com/alibaba/nacos/core/remote/SslContextChangeAware.java Normal file
View File

@ -0,0 +1,43 @@
/*
* Copyright 1999-2020 Alibaba Group Holding Ltd.
*
* 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 com.alibaba.nacos.core.remote;
/**
* ssl context refresher spi holder.
*
* @author liuzunfei
* @version $Id: RequestFilters.java, v 0.1 2023年03月17日 12:00 PM liuzunfei Exp $
*/
public interface SslContextChangeAware {
/**
* init rpc server ssl context.
*
* @param baseRpcServer rpc server.
*/
void init(BaseRpcServer baseRpcServer);
/**
* do something on ssl context change.
*/
void onSslContextChange();
/**
* shutdown to clear context.
*/
void shutdown();
}

85
core/src/main/java/com/alibaba/nacos/core/remote/grpc/BaseGrpcServer.java
View File

@ -22,10 +22,12 @@ import com.alibaba.nacos.common.packagescan.resource.Resource;
import com.alibaba.nacos.common.packagescan.resource.ResourceLoader;
import com.alibaba.nacos.common.remote.ConnectionType;
import com.alibaba.nacos.common.utils.JacksonUtils;
import com.alibaba.nacos.common.utils.StringUtils;
import com.alibaba.nacos.common.utils.TlsTypeResolve;
import com.alibaba.nacos.core.remote.BaseRpcServer;
import com.alibaba.nacos.core.remote.ConnectionManager;
import com.alibaba.nacos.core.utils.Loggers;
import com.alibaba.nacos.sys.env.EnvUtil;
import io.grpc.CompressorRegistry;
import io.grpc.DecompressorRegistry;
@ -75,6 +77,8 @@ public abstract class BaseGrpcServer extends BaseRpcServer {
@Autowired
private ConnectionManager connectionManager;
private OptionalTlsProtocolNegotiator optionalTlsProtocolNegotiator;
@Override
public ConnectionType getConnectionType() {
return ConnectionType.GRPC;
@ -86,12 +90,10 @@ public abstract class BaseGrpcServer extends BaseRpcServer {
addServices(handlerRegistry, new GrpcConnectionInterceptor());
NettyServerBuilder builder = NettyServerBuilder.forPort(getServicePort()).executor(getRpcExecutor());
if (grpcServerConfig.getEnableTls()) {
if (grpcServerConfig.getCompatibility()) {
builder.protocolNegotiator(new OptionalTlsProtocolNegotiator(getSslContextBuilder()));
} else {
builder.sslContext(getSslContextBuilder());
}
if (rpcServerTlsConfig.getEnableTls()) {
builder.protocolNegotiator(
new OptionalTlsProtocolNegotiator(getSslContextBuilder(), rpcServerTlsConfig.getCompatibility()));
}
server = builder.maxInboundMessageSize(getMaxInboundMessageSize()).fallbackHandlerRegistry(handlerRegistry)
@ -100,12 +102,27 @@ public abstract class BaseGrpcServer extends BaseRpcServer {
.addTransportFilter(new AddressTransportFilter(connectionManager))
.keepAliveTime(getKeepAliveTime(), TimeUnit.MILLISECONDS)
.keepAliveTimeout(getKeepAliveTimeout(), TimeUnit.MILLISECONDS)
.permitKeepAliveTime(getPermitKeepAliveTime(), TimeUnit.MILLISECONDS)
.build();
.permitKeepAliveTime(getPermitKeepAliveTime(), TimeUnit.MILLISECONDS).build();
server.start();
}
/**
* reload ssl context.
*/
public void reloadSslContext() {
if (optionalTlsProtocolNegotiator != null) {
try {
optionalTlsProtocolNegotiator.setSslContext(getSslContextBuilder());
} catch (Throwable throwable) {
Loggers.REMOTE.info("Nacos {} Rpc server reload ssl context fail at port {} and tls config:{}",
this.getClass().getSimpleName(), getServicePort(),
JacksonUtils.toJson(super.rpcServerTlsConfig));
throw throwable;
}
}
}
protected long getPermitKeepAliveTime() {
return GrpcServerConstants.GrpcConfig.DEFAULT_GRPC_PERMIT_KEEP_ALIVE_TIME;
}
@ -131,18 +148,17 @@ public abstract class BaseGrpcServer extends BaseRpcServer {
// unary common call register.
final MethodDescriptor<Payload, Payload> unaryPayloadMethod = MethodDescriptor.<Payload, Payload>newBuilder()
.setType(MethodDescriptor.MethodType.UNARY)
.setFullMethodName(MethodDescriptor.generateFullMethodName(GrpcServerConstants.REQUEST_SERVICE_NAME,
.setType(MethodDescriptor.MethodType.UNARY).setFullMethodName(
MethodDescriptor.generateFullMethodName(GrpcServerConstants.REQUEST_SERVICE_NAME,
GrpcServerConstants.REQUEST_METHOD_NAME))
.setRequestMarshaller(ProtoUtils.marshaller(Payload.getDefaultInstance()))
.setResponseMarshaller(ProtoUtils.marshaller(Payload.getDefaultInstance())).build();
final ServerCallHandler<Payload, Payload> payloadHandler = ServerCalls
.asyncUnaryCall((request, responseObserver) -> grpcCommonRequestAcceptor.request(request, responseObserver));
final ServerCallHandler<Payload, Payload> payloadHandler = ServerCalls.asyncUnaryCall(
(request, responseObserver) -> grpcCommonRequestAcceptor.request(request, responseObserver));
final ServerServiceDefinition serviceDefOfUnaryPayload = ServerServiceDefinition.builder(
GrpcServerConstants.REQUEST_SERVICE_NAME)
.addMethod(unaryPayloadMethod, payloadHandler).build();
GrpcServerConstants.REQUEST_SERVICE_NAME).addMethod(unaryPayloadMethod, payloadHandler).build();
handlerRegistry.addService(ServerInterceptors.intercept(serviceDefOfUnaryPayload, serverInterceptor));
// bi stream register.
@ -150,14 +166,14 @@ public abstract class BaseGrpcServer extends BaseRpcServer {
(responseObserver) -> grpcBiStreamRequestAcceptor.requestBiStream(responseObserver));
final MethodDescriptor<Payload, Payload> biStreamMethod = MethodDescriptor.<Payload, Payload>newBuilder()
.setType(MethodDescriptor.MethodType.BIDI_STREAMING).setFullMethodName(MethodDescriptor
.generateFullMethodName(GrpcServerConstants.REQUEST_BI_STREAM_SERVICE_NAME,
.setType(MethodDescriptor.MethodType.BIDI_STREAMING).setFullMethodName(
MethodDescriptor.generateFullMethodName(GrpcServerConstants.REQUEST_BI_STREAM_SERVICE_NAME,
GrpcServerConstants.REQUEST_BI_STREAM_METHOD_NAME))
.setRequestMarshaller(ProtoUtils.marshaller(Payload.newBuilder().build()))
.setResponseMarshaller(ProtoUtils.marshaller(Payload.getDefaultInstance())).build();
final ServerServiceDefinition serviceDefOfBiStream = ServerServiceDefinition
.builder(GrpcServerConstants.REQUEST_BI_STREAM_SERVICE_NAME).addMethod(biStreamMethod, biStreamHandler).build();
final ServerServiceDefinition serviceDefOfBiStream = ServerServiceDefinition.builder(
GrpcServerConstants.REQUEST_BI_STREAM_SERVICE_NAME).addMethod(biStreamMethod, biStreamHandler).build();
handlerRegistry.addService(ServerInterceptors.intercept(serviceDefOfBiStream, serverInterceptor));
}
@ -171,37 +187,40 @@ public abstract class BaseGrpcServer extends BaseRpcServer {
private SslContext getSslContextBuilder() {
try {
if (StringUtils.isBlank(grpcServerConfig.getCertChainFile()) || StringUtils.isBlank(grpcServerConfig.getCertPrivateKey())) {
if (StringUtils.isBlank(rpcServerTlsConfig.getCertChainFile()) || StringUtils.isBlank(
rpcServerTlsConfig.getCertPrivateKey())) {
throw new IllegalArgumentException("Server certChainFile or certPrivateKey must be not null");
}
InputStream certificateChainFile = getInputStream(grpcServerConfig.getCertChainFile(), "certChainFile");
InputStream privateKeyFile = getInputStream(grpcServerConfig.getCertPrivateKey(), "certPrivateKey");
SslContextBuilder sslClientContextBuilder = SslContextBuilder.forServer(certificateChainFile, privateKeyFile,
grpcServerConfig.getCertPrivateKeyPassword());
InputStream certificateChainFile = getInputStream(rpcServerTlsConfig.getCertChainFile(), "certChainFile");
InputStream privateKeyFile = getInputStream(rpcServerTlsConfig.getCertPrivateKey(), "certPrivateKey");
SslContextBuilder sslClientContextBuilder = SslContextBuilder.forServer(certificateChainFile,
privateKeyFile, rpcServerTlsConfig.getCertPrivateKeyPassword());
if (StringUtils.isNotBlank(grpcServerConfig.getProtocols())) {
sslClientContextBuilder.protocols(grpcServerConfig.getProtocols().split(","));
if (StringUtils.isNotBlank(rpcServerTlsConfig.getProtocols())) {
sslClientContextBuilder.protocols(rpcServerTlsConfig.getProtocols().split(","));
}
if (StringUtils.isNotBlank(grpcServerConfig.getCiphers())) {
sslClientContextBuilder.ciphers(Arrays.asList(grpcServerConfig.getCiphers().split(",")));
if (StringUtils.isNotBlank(rpcServerTlsConfig.getCiphers())) {
sslClientContextBuilder.ciphers(Arrays.asList(rpcServerTlsConfig.getCiphers().split(",")));
}
if (grpcServerConfig.getMutualAuthEnable()) {
if (rpcServerTlsConfig.getMutualAuthEnable()) {
// trust all certificate
if (grpcServerConfig.getTrustAll()) {
if (rpcServerTlsConfig.getTrustAll()) {
sslClientContextBuilder.trustManager(InsecureTrustManagerFactory.INSTANCE);
} else {
if (StringUtils.isBlank(grpcServerConfig.getTrustCollectionCertFile())) {
throw new IllegalArgumentException("enable mutual auth,trustCollectionCertFile must be not null");
if (StringUtils.isBlank(rpcServerTlsConfig.getTrustCollectionCertFile())) {
throw new IllegalArgumentException(
"enable mutual auth,trustCollectionCertFile must be not null");
}
InputStream clientCert = getInputStream(grpcServerConfig.getTrustCollectionCertFile(), "trustCollectionCertFile");
InputStream clientCert = getInputStream(rpcServerTlsConfig.getTrustCollectionCertFile(),
"trustCollectionCertFile");
sslClientContextBuilder.trustManager(clientCert);
}
sslClientContextBuilder.clientAuth(ClientAuth.REQUIRE);
}
SslContextBuilder configure = GrpcSslContexts.configure(sslClientContextBuilder,
TlsTypeResolve.getSslProvider(grpcServerConfig.getSslProvider()));
TlsTypeResolve.getSslProvider(rpcServerTlsConfig.getSslProvider()));
return configure.build();
} catch (SSLException e) {
throw new RuntimeException(e);

21
core/src/main/java/com/alibaba/nacos/core/remote/grpc/OptionalTlsProtocolNegotiator.java
View File

@ -40,9 +40,16 @@ public class OptionalTlsProtocolNegotiator implements InternalProtocolNegotiator
private static final int MAGIC_VALUE = 5;
private boolean supportPlainText;
private SslContext sslContext;
public OptionalTlsProtocolNegotiator(SslContext sslContext) {
public OptionalTlsProtocolNegotiator(SslContext sslContext, boolean supportPlainText) {
this.sslContext = sslContext;
this.supportPlainText = supportPlainText;
}
void setSslContext(SslContext sslContext) {
this.sslContext = sslContext;
}
@ -53,10 +60,8 @@ public class OptionalTlsProtocolNegotiator implements InternalProtocolNegotiator
@Override
public ChannelHandler newHandler(GrpcHttp2ConnectionHandler grpcHttp2ConnectionHandler) {
ChannelHandler plaintext =
InternalProtocolNegotiators.serverPlaintext().newHandler(grpcHttp2ConnectionHandler);
ChannelHandler ssl =
InternalProtocolNegotiators.serverTls(sslContext).newHandler(grpcHttp2ConnectionHandler);
ChannelHandler plaintext = InternalProtocolNegotiators.serverPlaintext().newHandler(grpcHttp2ConnectionHandler);
ChannelHandler ssl = InternalProtocolNegotiators.serverTls(sslContext).newHandler(grpcHttp2ConnectionHandler);
ChannelHandler decoder = new PortUnificationServerHandler(ssl, plaintext);
return decoder;
}
@ -79,6 +84,7 @@ public class OptionalTlsProtocolNegotiator implements InternalProtocolNegotiator
}
public class PortUnificationServerHandler extends ByteToMessageDecoder {
private ProtocolNegotiationEvent pne;
private final ChannelHandler ssl;
@ -96,12 +102,11 @@ public class OptionalTlsProtocolNegotiator implements InternalProtocolNegotiator
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
throws Exception {
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
if (in.readableBytes() < MAGIC_VALUE) {
return;
}
if (isSsl(in)) {
if (isSsl(in) || !supportPlainText) {
ctx.pipeline().addAfter(ctx.name(), (String) null, this.ssl);
ctx.fireUserEventTriggered(pne);
ctx.pipeline().remove(this);

31
core/src/test/java/com/alibaba/nacos/core/remote/grpc/GrpcServerTest.java
View File

@ -20,10 +20,14 @@ package com.alibaba.nacos.core.remote.grpc;
import com.alibaba.nacos.common.remote.ConnectionType;
import com.alibaba.nacos.core.remote.RpcServerTlsConfig;
import com.alibaba.nacos.sys.env.EnvUtil;
import com.alibaba.nacos.sys.utils.ApplicationUtils;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.mock.env.MockEnvironment;
@ -43,16 +47,25 @@ public class GrpcServerTest {
private final RpcServerTlsConfig grpcServerConfig = mock(RpcServerTlsConfig.class);
@Before
public void setUp() {
static MockedStatic<ApplicationUtils> applicationUtilsMockedStatic = null;
@BeforeClass
public static void setUp() {
EnvUtil.setEnvironment(new MockEnvironment());
applicationUtilsMockedStatic = Mockito.mockStatic(ApplicationUtils.class);
}
@AfterClass
public static void after() {
applicationUtilsMockedStatic.close();
}
@Test
public void testGrpcSdkServer() throws Exception {
BaseGrpcServer grpcSdkServer = new GrpcSdkServer();
grpcSdkServer.setGrpcServerConfig(grpcServerConfig);
grpcSdkServer.setRpcServerTlsConfig(grpcServerConfig);
when(grpcServerConfig.getEnableTls()).thenReturn(false);
when(ApplicationUtils.getBean(RpcServerTlsConfig.class)).thenReturn(grpcServerConfig);
grpcSdkServer.start();
Assert.assertEquals(grpcSdkServer.getConnectionType(), ConnectionType.GRPC);
Assert.assertEquals(grpcSdkServer.rpcPortOffset(), 1000);
@ -62,8 +75,9 @@ public class GrpcServerTest {
@Test
public void testGrpcClusterServer() throws Exception {
BaseGrpcServer grpcSdkServer = new GrpcClusterServer();
grpcSdkServer.setGrpcServerConfig(grpcServerConfig);
grpcSdkServer.setRpcServerTlsConfig(grpcServerConfig);
when(grpcServerConfig.getEnableTls()).thenReturn(false);
when(ApplicationUtils.getBean(RpcServerTlsConfig.class)).thenReturn(grpcServerConfig);
grpcSdkServer.start();
Assert.assertEquals(grpcSdkServer.getConnectionType(), ConnectionType.GRPC);
Assert.assertEquals(grpcSdkServer.rpcPortOffset(), 1001);
@ -89,7 +103,8 @@ public class GrpcServerTest {
when(grpcServerConfig.getCertPrivateKey()).thenReturn("test-server-key.pem");
when(grpcServerConfig.getCertChainFile()).thenReturn("test-server-cert.pem");
grpcSdkServer.setGrpcServerConfig(grpcServerConfig);
when(ApplicationUtils.getBean(RpcServerTlsConfig.class)).thenReturn(grpcServerConfig);
grpcSdkServer.setRpcServerTlsConfig(grpcServerConfig);
grpcSdkServer.start();
grpcSdkServer.shutdownServer();
}
@ -115,7 +130,7 @@ public class GrpcServerTest {
when(grpcServerConfig.getProtocols()).thenReturn("TLSv1.2,TLSv1.3");
when(grpcServerConfig.getCertPrivateKey()).thenReturn("test-server-key.pem");
when(grpcServerConfig.getCertChainFile()).thenReturn("test-server-cert.pem");
grpcSdkServer.setGrpcServerConfig(grpcServerConfig);
grpcSdkServer.setRpcServerTlsConfig(grpcServerConfig);
grpcSdkServer.start();
grpcSdkServer.shutdownServer();
}
@ -143,7 +158,7 @@ public class GrpcServerTest {
when(grpcServerConfig.getCertChainFile()).thenReturn("test-server-cert.pem");
when(grpcServerConfig.getTrustCollectionCertFile()).thenReturn("test-ca-cert.pem");
grpcSdkServer.setGrpcServerConfig(grpcServerConfig);
grpcSdkServer.setRpcServerTlsConfig(grpcServerConfig);
grpcSdkServer.start();
grpcSdkServer.shutdownServer();