Exact to support dynamic cert (#11578)

.licenserc.yaml

@@ -67,6 +67,7 @@ header:
     - 'CNAME'
     - 'Jenkinsfile'
     - '**/vendor/**'
+    - '**/src/test/resources/certs/**'
     - 'dubbo-common/src/main/java/org/apache/dubbo/common/threadlocal/InternalThreadLocal.java'
     - 'dubbo-common/src/main/java/org/apache/dubbo/common/threadlocal/InternalThreadLocalMap.java'
     - 'dubbo-common/src/main/java/org/apache/dubbo/common/timer/HashedWheelTimer.java'

codestyle/checkstyle-suppressions.xml

@@ -5,5 +5,6 @@
 <suppressions>
     <suppress files="[\\/]src[\\/]main[\\/]java[\\/]com[\\/]alibaba[\\/]com[\\/]caucho[\\/]hessian" checks=".*"/>
     <suppress files="[\\/]build[\\/]generated[\\/]source[\\/]proto" checks=".*"/>
+    <suppress files="[\\/]target[\\/]generated-sources[\\/]protobuf" checks=".*"/>
     <suppress files="Yylex\.java" checks="AvoidEscapedUnicodeCharacters"/>
 </suppressions>

dubbo-common/src/main/java/org/apache/dubbo/common/CommonScopeModelInitializer.java

@@ -20,6 +20,7 @@ import org.apache.dubbo.common.beans.factory.ScopeBeanFactory;
 import org.apache.dubbo.common.config.ConfigurationCache;
 import org.apache.dubbo.common.convert.ConverterUtil;
 import org.apache.dubbo.common.lang.ShutdownHookCallbacks;
+import org.apache.dubbo.common.ssl.CertManager;
 import org.apache.dubbo.common.status.reporter.FrameworkStatusReportService;
 import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
 import org.apache.dubbo.common.utils.DefaultSerializeClassChecker;
@@ -38,6 +39,7 @@ public class CommonScopeModelInitializer implements ScopeModelInitializer {
         beanFactory.registerBean(ConverterUtil.class);
         beanFactory.registerBean(SerializeSecurityManager.class);
         beanFactory.registerBean(DefaultSerializeClassChecker.class);
+        beanFactory.registerBean(CertManager.class);
     }
 
     @Override

dubbo-common/src/main/java/org/apache/dubbo/common/constants/LoggerCodeConstants.java

@@ -380,6 +380,12 @@ public interface LoggerCodeConstants {
 
     String CONFIG_DUBBO_BEAN_NOT_FOUND = "5-40";
 
+    String CONFIG_SSL_PATH_LOAD_FAILED = "5-41";
+
+    String CONFIG_SSL_CERT_GENERATE_FAILED = "5-42";
+
+    String CONFIG_SSL_CONNECT_INSECURE = "5-43";
+
     // Transport module
     String TRANSPORT_FAILED_CONNECT_PROVIDER = "6-1";
 

dubbo-common/src/main/java/org/apache/dubbo/common/ssl/AuthPolicy.java

@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.common.ssl;
+
+public enum AuthPolicy {
+    NONE,
+    SERVER_AUTH,
+    CLIENT_AUTH
+}

dubbo-common/src/main/java/org/apache/dubbo/common/ssl/Cert.java

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.common.ssl;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+
+public class Cert {
+    private final byte[] keyCertChain;
+    private final byte[] privateKey;
+    private final byte[] trustCert;
+
+    private final String password;
+
+    public Cert(byte[] keyCertChain, byte[] privateKey, byte[] trustCert) {
+        this(keyCertChain, privateKey, trustCert, null);
+    }
+
+    public Cert(byte[] keyCertChain, byte[] privateKey, byte[] trustCert, String password) {
+        this.keyCertChain = keyCertChain;
+        this.privateKey = privateKey;
+        this.trustCert = trustCert;
+        this.password = password;
+    }
+
+    public byte[] getKeyCertChain() {
+        return keyCertChain;
+    }
+
+    public InputStream getKeyCertChainInputStream() {
+        return keyCertChain != null ? new ByteArrayInputStream(keyCertChain) : null;
+    }
+
+    public byte[] getPrivateKey() {
+        return privateKey;
+    }
+
+    public InputStream getPrivateKeyInputStream() {
+        return privateKey != null ? new ByteArrayInputStream(privateKey) : null;
+    }
+
+    public byte[] getTrustCert() {
+        return trustCert;
+    }
+
+    public InputStream getTrustCertInputStream() {
+        return trustCert != null ? new ByteArrayInputStream(trustCert) : null;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+}

dubbo-common/src/main/java/org/apache/dubbo/common/ssl/CertManager.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.common.ssl;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+
+import java.net.SocketAddress;
+import java.util.List;
+
+public class CertManager {
+    private final List<CertProvider> certProviders;
+
+    public CertManager(FrameworkModel frameworkModel) {
+        this.certProviders = frameworkModel.getExtensionLoader(CertProvider.class).getActivateExtensions();
+    }
+
+    public ProviderCert getProviderConnectionConfig(URL localAddress, SocketAddress remoteAddress) {
+        for (CertProvider certProvider : certProviders) {
+            if (certProvider.isSupport(localAddress)) {
+                ProviderCert cert = certProvider.getProviderConnectionConfig(localAddress);
+                if (cert != null) {
+                    return cert;
+                }
+            }
+        }
+        return null;
+    }
+
+    public Cert getConsumerConnectionConfig(URL remoteAddress) {
+        for (CertProvider certProvider : certProviders) {
+            if (certProvider.isSupport(remoteAddress)) {
+                Cert cert = certProvider.getConsumerConnectionConfig(remoteAddress);
+                if (cert != null) {
+                    return cert;
+                }
+            }
+        }
+        return null;
+    }
+}

dubbo-common/src/main/java/org/apache/dubbo/common/ssl/CertProvider.java

@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.common.ssl;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.ExtensionScope;
+import org.apache.dubbo.common.extension.SPI;
+
+@SPI(scope = ExtensionScope.FRAMEWORK)
+public interface CertProvider {
+    boolean isSupport(URL address);
+
+    ProviderCert getProviderConnectionConfig(URL localAddress);
+
+    Cert getConsumerConnectionConfig(URL remoteAddress);
+}

dubbo-common/src/main/java/org/apache/dubbo/common/ssl/ProviderCert.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.common.ssl;
+
+public class ProviderCert extends Cert {
+    private final AuthPolicy authPolicy;
+
+    public ProviderCert(byte[] keyCertChain, byte[] privateKey, byte[] trustCert, AuthPolicy authPolicy) {
+        super(keyCertChain, privateKey, trustCert);
+        this.authPolicy = authPolicy;
+    }
+
+    public ProviderCert(byte[] keyCertChain, byte[] privateKey, byte[] trustCert, String password, AuthPolicy authPolicy) {
+        super(keyCertChain, privateKey, trustCert, password);
+        this.authPolicy = authPolicy;
+    }
+
+    public AuthPolicy getAuthPolicy() {
+        return authPolicy;
+    }
+}

dubbo-common/src/main/java/org/apache/dubbo/common/ssl/impl/SSLConfigCertProvider.java

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.common.ssl.impl;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.constants.LoggerCodeConstants;
+import org.apache.dubbo.common.extension.Activate;
+import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.ssl.AuthPolicy;
+import org.apache.dubbo.common.ssl.Cert;
+import org.apache.dubbo.common.ssl.CertProvider;
+import org.apache.dubbo.common.ssl.ProviderCert;
+import org.apache.dubbo.common.utils.IOUtils;
+
+import java.io.IOException;
+import java.util.Objects;
+
+@Activate(order = Integer.MAX_VALUE - 10000)
+public class SSLConfigCertProvider implements CertProvider {
+    private final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(SSLConfigCertProvider.class);
+
+    @Override
+    public boolean isSupport(URL address) {
+        return address.getOrDefaultApplicationModel().getApplicationConfigManager().getSsl()
+            .isPresent();
+    }
+
+    @Override
+    public ProviderCert getProviderConnectionConfig(URL localAddress) {
+        return localAddress.getOrDefaultApplicationModel().getApplicationConfigManager().getSsl()
+            .filter(sslConfig -> Objects.nonNull(sslConfig.getServerKeyCertChainPath()))
+            .filter(sslConfig -> Objects.nonNull(sslConfig.getServerPrivateKeyPath()))
+            .map(sslConfig -> {
+                try {
+                    return new ProviderCert(
+                        IOUtils.toByteArray(sslConfig.getServerKeyCertChainPathStream()),
+                        IOUtils.toByteArray(sslConfig.getServerPrivateKeyPathStream()),
+                        sslConfig.getServerTrustCertCollectionPath() != null ? IOUtils.toByteArray(sslConfig.getServerTrustCertCollectionPathStream()) : null,
+                        sslConfig.getServerKeyPassword(), AuthPolicy.CLIENT_AUTH);
+                } catch (IOException e) {
+                    logger.warn(LoggerCodeConstants.CONFIG_SSL_PATH_LOAD_FAILED, "", "", "Failed to load ssl config.", e);
+                    return null;
+                }
+            }).orElse(null);
+    }
+
+    @Override
+    public Cert getConsumerConnectionConfig(URL remoteAddress) {
+        return remoteAddress.getOrDefaultApplicationModel().getApplicationConfigManager().getSsl()
+            .filter(sslConfig -> Objects.nonNull(sslConfig.getClientKeyCertChainPath()))
+            .filter(sslConfig -> Objects.nonNull(sslConfig.getClientPrivateKeyPath()))
+            .map(sslConfig -> {
+                try {
+                    return new Cert(
+                        IOUtils.toByteArray(sslConfig.getClientKeyCertChainPathStream()),
+                        IOUtils.toByteArray(sslConfig.getClientPrivateKeyPathStream()),
+                        sslConfig.getClientTrustCertCollectionPath() != null ? IOUtils.toByteArray(sslConfig.getClientTrustCertCollectionPathStream()) : null,
+                        sslConfig.getClientKeyPassword());
+                } catch (IOException e) {
+                    logger.warn(LoggerCodeConstants.CONFIG_SSL_PATH_LOAD_FAILED, "", "", "Failed to load ssl config.", e);
+                    return null;
+                }
+            }).orElse(null);
+    }
+}

dubbo-common/src/main/java/org/apache/dubbo/common/utils/IOUtils.java

@@ -19,6 +19,7 @@ package org.apache.dubbo.common.utils;
 import org.apache.dubbo.common.constants.CommonConstants;
 
 import java.io.BufferedReader;
+import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
@@ -277,4 +278,15 @@ public class IOUtils {
             }
         }
     }
+
+    public static byte[] toByteArray(final InputStream inputStream) throws IOException {
+        try (final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()) {
+            byte[] buffer = new byte[1024];
+            int n;
+            while (-1 != (n = inputStream.read(buffer))) {
+                byteArrayOutputStream.write(buffer, 0, n);
+            }
+            return byteArrayOutputStream.toByteArray();
+        }
+    }
 }

dubbo-common/src/main/java/org/apache/dubbo/config/SslConfig.java

@@ -64,6 +64,11 @@ public class SslConfig extends AbstractConfig {
     private InputStream clientPrivateKeyPathStream;
     private InputStream clientTrustCertCollectionPathStream;
 
+    private String caAddress;
+    private String envType;
+    private String caCertPath;
+    private String oidcTokenPath;
+
     public SslConfig() {
     }
 
@@ -143,6 +148,38 @@ public class SslConfig extends AbstractConfig {
         this.clientTrustCertCollectionPath = clientTrustCertCollectionPath;
     }
 
+    public String getCaAddress() {
+        return caAddress;
+    }
+
+    public void setCaAddress(String caAddress) {
+        this.caAddress = caAddress;
+    }
+
+    public String getEnvType() {
+        return envType;
+    }
+
+    public void setEnvType(String envType) {
+        this.envType = envType;
+    }
+
+    public String getCaCertPath() {
+        return caCertPath;
+    }
+
+    public void setCaCertPath(String caCertPath) {
+        this.caCertPath = caCertPath;
+    }
+
+    public String getOidcTokenPath() {
+        return oidcTokenPath;
+    }
+
+    public void setOidcTokenPath(String oidcTokenPath) {
+        this.oidcTokenPath = oidcTokenPath;
+    }
+
     @Transient
     public InputStream getServerKeyCertChainPathStream() throws IOException {
         if (serverKeyCertChainPath != null) {

dubbo-common/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.common.ssl.CertProvider

@@ -0,0 +1 @@
+ssl-config=org.apache.dubbo.common.ssl.impl.SSLConfigCertProvider

dubbo-common/src/test/java/org/apache/dubbo/common/ssl/CertManagerTest.java

@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.common.ssl;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+class CertManagerTest {
+    private FrameworkModel frameworkModel;
+    private URL url;
+    @BeforeEach
+    void setup() {
+        FirstCertProvider.setProviderCert(null);
+        FirstCertProvider.setCert(null);
+        FirstCertProvider.setSupport(false);
+
+        SecondCertProvider.setProviderCert(null);
+        SecondCertProvider.setCert(null);
+        SecondCertProvider.setSupport(false);
+
+        frameworkModel = new FrameworkModel();
+        url = URL.valueOf("dubbo://").setScopeModel(frameworkModel.newApplication());
+    }
+
+    @AfterEach
+    void teardown() {
+        frameworkModel.destroy();
+    }
+
+    @Test
+    void testGetConsumerConnectionConfig() {
+        CertManager certManager = new CertManager(frameworkModel);
+
+        Assertions.assertNull(certManager.getConsumerConnectionConfig(url));
+
+        Cert cert1 = Mockito.mock(Cert.class);
+        FirstCertProvider.setCert(cert1);
+        Assertions.assertNull(certManager.getConsumerConnectionConfig(url));
+
+        FirstCertProvider.setSupport(true);
+        Assertions.assertEquals(cert1, certManager.getConsumerConnectionConfig(url));
+
+        Cert cert2 = Mockito.mock(Cert.class);
+        SecondCertProvider.setCert(cert2);
+        Assertions.assertEquals(cert1, certManager.getConsumerConnectionConfig(url));
+
+        SecondCertProvider.setSupport(true);
+        Assertions.assertEquals(cert1, certManager.getConsumerConnectionConfig(url));
+
+        FirstCertProvider.setSupport(false);
+        Assertions.assertEquals(cert2, certManager.getConsumerConnectionConfig(url));
+
+        FirstCertProvider.setSupport(true);
+        FirstCertProvider.setCert(null);
+        Assertions.assertEquals(cert2, certManager.getConsumerConnectionConfig(url));
+    }
+
+    @Test
+    void testGetProviderConnectionConfig() {
+        CertManager certManager = new CertManager(frameworkModel);
+
+        Assertions.assertNull(certManager.getProviderConnectionConfig(url, null));
+
+        ProviderCert providerCert1 = Mockito.mock(ProviderCert.class);
+        FirstCertProvider.setProviderCert(providerCert1);
+        Assertions.assertNull(certManager.getProviderConnectionConfig(url, null));
+
+        FirstCertProvider.setSupport(true);
+        Assertions.assertEquals(providerCert1, certManager.getProviderConnectionConfig(url, null));
+
+        ProviderCert providerCert2 = Mockito.mock(ProviderCert.class);
+        SecondCertProvider.setProviderCert(providerCert2);
+        Assertions.assertEquals(providerCert1, certManager.getProviderConnectionConfig(url, null));
+
+        SecondCertProvider.setSupport(true);
+        Assertions.assertEquals(providerCert1, certManager.getProviderConnectionConfig(url, null));
+
+        FirstCertProvider.setSupport(false);
+        Assertions.assertEquals(providerCert2, certManager.getProviderConnectionConfig(url, null));
+
+        FirstCertProvider.setSupport(true);
+        FirstCertProvider.setProviderCert(null);
+        Assertions.assertEquals(providerCert2, certManager.getProviderConnectionConfig(url, null));
+    }
+}

dubbo-common/src/test/java/org/apache/dubbo/common/ssl/FirstCertProvider.java

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.common.ssl;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.Activate;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+@Activate(order = -10000)
+public class FirstCertProvider implements CertProvider {
+    private static final AtomicBoolean isSupport = new AtomicBoolean(false);
+    private static final AtomicReference<ProviderCert> providerCert = new AtomicReference<>();
+    private static final AtomicReference<Cert> cert = new AtomicReference<>();
+    @Override
+    public boolean isSupport(URL address) {
+        return isSupport.get();
+    }
+
+    @Override
+    public ProviderCert getProviderConnectionConfig(URL localAddress) {
+        return providerCert.get();
+    }
+
+    @Override
+    public Cert getConsumerConnectionConfig(URL remoteAddress) {
+        return cert.get();
+    }
+
+    public static void setSupport(boolean support) {
+        isSupport.set(support);
+    }
+
+    public static void setProviderCert(ProviderCert providerCert) {
+        FirstCertProvider.providerCert.set(providerCert);
+    }
+
+    public static void setCert(Cert cert) {
+        FirstCertProvider.cert.set(cert);
+    }
+}

dubbo-common/src/test/java/org/apache/dubbo/common/ssl/SSLConfigCertProviderTest.java

@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.common.ssl;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.ssl.impl.SSLConfigCertProvider;
+import org.apache.dubbo.common.utils.IOUtils;
+import org.apache.dubbo.config.SslConfig;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+class SSLConfigCertProviderTest {
+    @Test
+    void testSupported() {
+        FrameworkModel frameworkModel = new FrameworkModel();
+        ApplicationModel applicationModel = frameworkModel.newApplication();
+        SSLConfigCertProvider sslConfigCertProvider = new SSLConfigCertProvider();
+
+        URL url = URL.valueOf("").setScopeModel(applicationModel);
+        Assertions.assertFalse(sslConfigCertProvider.isSupport(url));
+
+        SslConfig sslConfig = new SslConfig();
+        applicationModel.getApplicationConfigManager().setSsl(sslConfig);
+        Assertions.assertTrue(sslConfigCertProvider.isSupport(url));
+
+        frameworkModel.destroy();
+    }
+
+    @Test
+    void testGetProviderConnectionConfig() throws IOException {
+        FrameworkModel frameworkModel = new FrameworkModel();
+        ApplicationModel applicationModel = frameworkModel.newApplication();
+        SSLConfigCertProvider sslConfigCertProvider = new SSLConfigCertProvider();
+
+        URL url = URL.valueOf("").setScopeModel(applicationModel);
+        Assertions.assertNull(sslConfigCertProvider.getProviderConnectionConfig(url));
+
+        SslConfig sslConfig = new SslConfig();
+        sslConfig.setServerKeyCertChainPath("keyCert");
+        sslConfig.setServerPrivateKeyPath("private");
+        applicationModel.getApplicationConfigManager().setSsl(sslConfig);
+        ProviderCert providerCert = sslConfigCertProvider.getProviderConnectionConfig(url);
+        Assertions.assertNull(providerCert);
+
+        sslConfig.setServerKeyCertChainPath(this.getClass().getClassLoader().getResource("certs/cert.pem").getFile());
+        sslConfig.setServerPrivateKeyPath(this.getClass().getClassLoader().getResource("certs/key.pem").getFile());
+        providerCert = sslConfigCertProvider.getProviderConnectionConfig(url);
+        Assertions.assertNotNull(providerCert);
+        Assertions.assertArrayEquals(IOUtils.toByteArray(this.getClass().getClassLoader().getResourceAsStream("certs/cert.pem")),
+                providerCert.getKeyCertChain());
+        Assertions.assertArrayEquals(IOUtils.toByteArray(this.getClass().getClassLoader().getResourceAsStream("certs/key.pem")),
+                providerCert.getPrivateKey());
+        Assertions.assertNull(providerCert.getTrustCert());
+
+        sslConfig.setServerTrustCertCollectionPath(this.getClass().getClassLoader().getResource("certs/ca.pem").getFile());
+        providerCert = sslConfigCertProvider.getProviderConnectionConfig(url);
+        Assertions.assertNotNull(providerCert);
+        Assertions.assertArrayEquals(IOUtils.toByteArray(this.getClass().getClassLoader().getResourceAsStream("certs/cert.pem")),
+                providerCert.getKeyCertChain());
+        Assertions.assertArrayEquals(IOUtils.toByteArray(this.getClass().getClassLoader().getResourceAsStream("certs/key.pem")),
+                providerCert.getPrivateKey());
+        Assertions.assertArrayEquals(IOUtils.toByteArray(this.getClass().getClassLoader().getResourceAsStream("certs/ca.pem")),
+                providerCert.getTrustCert());
+
+        frameworkModel.destroy();
+    }
+
+    @Test
+    void testGetConsumerConnectionConfig() throws IOException {
+        FrameworkModel frameworkModel = new FrameworkModel();
+        ApplicationModel applicationModel = frameworkModel.newApplication();
+        SSLConfigCertProvider sslConfigCertProvider = new SSLConfigCertProvider();
+
+        URL url = URL.valueOf("").setScopeModel(applicationModel);
+        Assertions.assertNull(sslConfigCertProvider.getConsumerConnectionConfig(url));
+
+        SslConfig sslConfig = new SslConfig();
+        sslConfig.setClientKeyCertChainPath("keyCert");
+        sslConfig.setClientPrivateKeyPath("private");
+        applicationModel.getApplicationConfigManager().setSsl(sslConfig);
+        Cert cert = sslConfigCertProvider.getConsumerConnectionConfig(url);
+        Assertions.assertNull(cert);
+
+        sslConfig.setClientKeyCertChainPath(this.getClass().getClassLoader().getResource("certs/cert.pem").getFile());
+        sslConfig.setClientPrivateKeyPath(this.getClass().getClassLoader().getResource("certs/key.pem").getFile());
+        cert = sslConfigCertProvider.getConsumerConnectionConfig(url);
+        Assertions.assertNotNull(cert);
+        Assertions.assertArrayEquals(IOUtils.toByteArray(this.getClass().getClassLoader().getResourceAsStream("certs/cert.pem")),
+                cert.getKeyCertChain());
+        Assertions.assertArrayEquals(IOUtils.toByteArray(this.getClass().getClassLoader().getResourceAsStream("certs/key.pem")),
+                cert.getPrivateKey());
+
+        sslConfig.setClientTrustCertCollectionPath(this.getClass().getClassLoader().getResource("certs/ca.pem").getFile());
+        cert = sslConfigCertProvider.getConsumerConnectionConfig(url);
+        Assertions.assertNotNull(cert);
+        Assertions.assertArrayEquals(IOUtils.toByteArray(this.getClass().getClassLoader().getResourceAsStream("certs/cert.pem")),
+                cert.getKeyCertChain());
+        Assertions.assertArrayEquals(IOUtils.toByteArray(this.getClass().getClassLoader().getResourceAsStream("certs/key.pem")),
+                cert.getPrivateKey());
+        Assertions.assertArrayEquals(IOUtils.toByteArray(this.getClass().getClassLoader().getResourceAsStream("certs/ca.pem")),
+                cert.getTrustCert());
+
+        frameworkModel.destroy();
+    }
+}

dubbo-common/src/test/java/org/apache/dubbo/common/ssl/SecondCertProvider.java

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.common.ssl;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.Activate;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+@Activate(order = 10000)
+public class SecondCertProvider implements CertProvider {
+    private static final AtomicBoolean isSupport = new AtomicBoolean(false);
+    private static final AtomicReference<ProviderCert> providerCert = new AtomicReference<>();
+    private static final AtomicReference<Cert> cert = new AtomicReference<>();
+    @Override
+    public boolean isSupport(URL address) {
+        return isSupport.get();
+    }
+
+    @Override
+    public ProviderCert getProviderConnectionConfig(URL localAddress) {
+        return providerCert.get();
+    }
+
+    @Override
+    public Cert getConsumerConnectionConfig(URL remoteAddress) {
+        return cert.get();
+    }
+
+    public static void setSupport(boolean support) {
+        isSupport.set(support);
+    }
+
+    public static void setProviderCert(ProviderCert providerCert) {
+        SecondCertProvider.providerCert.set(providerCert);
+    }
+
+    public static void setCert(Cert cert) {
+        SecondCertProvider.cert.set(cert);
+    }
+}

dubbo-common/src/test/resources/META-INF/dubbo/internal/org.apache.dubbo.common.ssl.CertProvider

@@ -0,0 +1,2 @@
+first=org.apache.dubbo.common.ssl.FirstCertProvider
+second=org.apache.dubbo.common.ssl.SecondCertProvider

dubbo-common/src/test/resources/certs/ca.pem

@@ -0,0 +1,15 @@
+-----BEGIN CERTIFICATE-----
+MIICSjCCAbOgAwIBAgIJAJHGGR4dGioHMA0GCSqGSIb3DQEBCwUAMFYxCzAJBgNV
+BAYTAkFVMRMwEQYDVQQIEwpTb21lLVN0YXRlMSEwHwYDVQQKExhJbnRlcm5ldCBX
+aWRnaXRzIFB0eSBMdGQxDzANBgNVBAMTBnRlc3RjYTAeFw0xNDExMTEyMjMxMjla
+Fw0yNDExMDgyMjMxMjlaMFYxCzAJBgNVBAYTAkFVMRMwEQYDVQQIEwpTb21lLVN0
+YXRlMSEwHwYDVQQKExhJbnRlcm5ldCBXaWRnaXRzIFB0eSBMdGQxDzANBgNVBAMT
+BnRlc3RjYTCBnzANBgkqhkiG9w0BAQEFAAOBjQAwgYkCgYEAwEDfBV5MYdlHVHJ7
++L4nxrZy7mBfAVXpOc5vMYztssUI7mL2/iYujiIXM+weZYNTEpLdjyJdu7R5gGUu
+g1jSVK/EPHfc74O7AyZU34PNIP4Sh33N+/A5YexrNgJlPY+E3GdVYi4ldWJjgkAd
+Qah2PH5ACLrIIC6tRka9hcaBlIECAwEAAaMgMB4wDAYDVR0TBAUwAwEB/zAOBgNV
+HQ8BAf8EBAMCAgQwDQYJKoZIhvcNAQELBQADgYEAHzC7jdYlzAVmddi/gdAeKPau
+sPBG/C2HCWqHzpCUHcKuvMzDVkY/MP2o6JIW2DBbY64bO/FceExhjcykgaYtCH/m
+oIU63+CFOTtR7otyQAWHqXa7q4SbCDlG7DyRFxqG0txPtGvy12lgldA2+RgcigQG
+Dfcog5wrJytaQ6UA0wE=
+-----END CERTIFICATE-----

dubbo-common/src/test/resources/certs/cert.pem

@@ -0,0 +1,18 @@
+-----BEGIN CERTIFICATE-----
+MIIC6TCCAlKgAwIBAgIBCjANBgkqhkiG9w0BAQsFADBWMQswCQYDVQQGEwJBVTET
+MBEGA1UECBMKU29tZS1TdGF0ZTEhMB8GA1UEChMYSW50ZXJuZXQgV2lkZ2l0cyBQ
+dHkgTHRkMQ8wDQYDVQQDEwZ0ZXN0Y2EwHhcNMTUxMTEwMDEwOTU4WhcNMjUxMTA3
+MDEwOTU4WjBaMQswCQYDVQQGEwJBVTETMBEGA1UECAwKU29tZS1TdGF0ZTEhMB8G
+A1UECgwYSW50ZXJuZXQgV2lkZ2l0cyBQdHkgTHRkMRMwEQYDVQQDDAp0ZXN0Y2xp
+ZW50MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDsVEfbob4W3lVCDLOVmx9K
+cdJnoZdvurGaTY87xNiopmaR8zCR7pFR9BX5L4bNG/PkuVLfVTVAKndyDCQggBBr
+UTaEITNbfWK9swHJEr20WnKfhS/wo/Xg5sqNNCrFRmnnnwOA4eDlvmYZEzSnJXV6
+pEro9bBH9uOCWWLqmaev7QIDAQABo4HCMIG/MAkGA1UdEwQCMAAwCwYDVR0PBAQD
+AgXgMB0GA1UdDgQWBBQAdbW5Vml/CnYwqdP3mOHDARU+8zBwBgNVHSMEaTBnoVqk
+WDBWMQswCQYDVQQGEwJBVTETMBEGA1UECBMKU29tZS1TdGF0ZTEhMB8GA1UEChMY
+SW50ZXJuZXQgV2lkZ2l0cyBQdHkgTHRkMQ8wDQYDVQQDEwZ0ZXN0Y2GCCQCRxhke
+HRoqBzAJBgNVHREEAjAAMAkGA1UdEgQCMAAwDQYJKoZIhvcNAQELBQADgYEAf4MM
+k+sdzd720DfrQ0PF2gDauR3M9uBubozDuMuF6ufAuQBJSKGQEGibXbUelrwHmnql
+UjTyfolVcxEBVaF4VFHmn7u6vP7S1NexIDdNUHcULqxIb7Tzl8JYq8OOHD2rQy4H
+s8BXaVIzw4YcaCGAMS0iDX052Sy7e2JhP8Noxvo=
+-----END CERTIFICATE-----

dubbo-common/src/test/resources/certs/key.pem

@@ -0,0 +1,16 @@
+-----BEGIN PRIVATE KEY-----
+MIICeQIBADANBgkqhkiG9w0BAQEFAASCAmMwggJfAgEAAoGBAOxUR9uhvhbeVUIM
+s5WbH0px0mehl2+6sZpNjzvE2KimZpHzMJHukVH0Ffkvhs0b8+S5Ut9VNUAqd3IM
+JCCAEGtRNoQhM1t9Yr2zAckSvbRacp+FL/Cj9eDmyo00KsVGaeefA4Dh4OW+ZhkT
+NKcldXqkSuj1sEf244JZYuqZp6/tAgMBAAECgYEAi2NSVqpZMafE5YYUTcMGe6QS
+k2jtpsqYgggI2RnLJ/2tNZwYI5pwP8QVSbnMaiF4gokD5hGdrNDfTnb2v+yIwYEH
+0w8+oG7Z81KodsiZSIDJfTGsAZhVNwOz9y0VD8BBZZ1/274Zh52AUKLjZS/ZwIbS
+W2ywya855dPnH/wj+0ECQQD9X8D920kByTNHhBG18biAEZ4pxs9f0OAG8333eVcI
+w2lJDLsYDZrCB2ocgA3lUdozlzPC7YDYw8reg0tkiRY5AkEA7sdNzOeQsQRn7++5
+0bP9DtT/iON1gbfxRzCfCfXdoOtfQWIzTePWtURt9X/5D9NofI0Rg5W2oGy/MLe5
+/sXHVQJBAIup5XrJDkQywNZyAUU2ecn2bCWBFjwtqd+LBmuMciI9fOKsZtEKZrz/
+U0lkeMRoSwvXE8wmGLjjrAbdfohrXFkCQQDZEx/LtIl6JINJQiswVe0tWr6k+ASP
+1WXoTm+HYpoF/XUvv9LccNF1IazFj34hwRQwhx7w/V52Ieb+p0jUMYGxAkEAjDhd
+9pBO1fKXWiXzi9ZKfoyTNcUq3eBSVKwPG2nItg5ycXengjT5sgcWDnciIzW7BIVI
+JiqOszq9GWESErAatg==
+-----END PRIVATE KEY-----

dubbo-distribution/dubbo-all/pom.xml

@@ -243,6 +243,13 @@
             <scope>compile</scope>
             <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-security</artifactId>
+            <version>${project.version}</version>
+            <scope>compile</scope>
+            <optional>true</optional>
+        </dependency>
         <dependency>
             <groupId>org.apache.dubbo</groupId>
             <artifactId>dubbo-reactive</artifactId>
@@ -515,6 +522,7 @@
                                     <include>org.apache.dubbo:dubbo-monitor-api</include>
                                     <include>org.apache.dubbo:dubbo-monitor-default</include>
                                     <include>org.apache.dubbo:dubbo-qos</include>
+                                    <include>org.apache.dubbo:dubbo-security</include>
                                     <include>org.apache.dubbo:dubbo-reactive</include>
                                     <include>org.apache.dubbo:dubbo-spring-security</include>
                                     <include>org.apache.dubbo:dubbo-registry-api</include>
@@ -898,6 +906,12 @@
                                         META-INF/dubbo/internal/org.apache.dubbo.registry.client.metadata.MetadataServiceURLBuilder
                                     </resource>
                                 </transformer>
+                                <transformer
+                                    implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
+                                    <resource>
+                                        META-INF/dubbo/internal/org.apache.dubbo.common.ssl.CertProvider
+                                    </resource>
+                                </transformer>
                                 <transformer
                                     implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                     <resource>

dubbo-distribution/dubbo-bom/pom.xml

@@ -299,6 +299,11 @@
                 <artifactId>dubbo-auth</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.apache.dubbo</groupId>
+                <artifactId>dubbo-security</artifactId>
+                <version>${project.version}</version>
+            </dependency>
             <dependency>
                 <groupId>org.apache.dubbo</groupId>
                 <artifactId>dubbo-qos</artifactId>

dubbo-maven-plugin/src/main/resources/META-INF/native-image/reflect-config.json

@@ -2599,6 +2599,36 @@
       }
     ]
   },
+  {
+    "name": "org.apache.dubbo.common.ssl.impl.SSLConfigCertProvider",
+    "allPublicMethods": true,
+    "methods": [
+      {
+        "name": "<init>",
+        "parameterTypes": []
+      }
+    ]
+  },
+  {
+    "name": "org.apache.dubbo.security.cert.DubboCertManager",
+    "allPublicMethods": true,
+    "methods": [
+      {
+        "name": "<init>",
+        "parameterTypes": ["org.apache.dubbo.rpc.model.FrameworkModel"]
+      }
+    ]
+  },
+  {
+    "name": "org.apache.dubbo.common.ssl.CertManager",
+    "allPublicMethods": true,
+    "methods": [
+      {
+        "name": "<init>",
+        "parameterTypes": ["org.apache.dubbo.rpc.model.FrameworkModel"]
+      }
+    ]
+  },
   {
     "name": "org.apache.dubbo.rpc.proxy.javassist.JavassistProxyFactory"
   },

dubbo-native-plugin/src/main/resources/META-INF/native-image/reflect-config.json

@@ -2599,6 +2599,36 @@
       }
     ]
   },
+  {
+    "name": "org.apache.dubbo.common.ssl.impl.SSLConfigCertProvider",
+    "allPublicMethods": true,
+    "methods": [
+      {
+        "name": "<init>",
+        "parameterTypes": []
+      }
+    ]
+  },
+  {
+    "name": "org.apache.dubbo.security.cert.DubboCertManager",
+    "allPublicMethods": true,
+    "methods": [
+      {
+        "name": "<init>",
+        "parameterTypes": ["org.apache.dubbo.rpc.model.FrameworkModel"]
+      }
+    ]
+  },
+  {
+    "name": "org.apache.dubbo.common.ssl.CertManager",
+    "allPublicMethods": true,
+    "methods": [
+      {
+        "name": "<init>",
+        "parameterTypes": ["org.apache.dubbo.rpc.model.FrameworkModel"]
+      }
+    ]
+  },
   {
     "name": "org.apache.dubbo.rpc.proxy.javassist.JavassistProxyFactory"
   },

dubbo-plugin/dubbo-security/pom.xml

@@ -0,0 +1,131 @@
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one or more
+  contributor license agreements.  See the NOTICE file distributed with
+  this work for additional information regarding copyright ownership.
+  The ASF licenses this file to You 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.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>org.apache.dubbo</groupId>
+        <artifactId>dubbo-plugin</artifactId>
+        <version>${revision}</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <properties>
+        <skip_maven_deploy>false</skip_maven_deploy>
+        <protobuf-java.version>3.21.12</protobuf-java.version>
+        <grpc.version>1.41.0</grpc.version>
+    </properties>
+
+    <artifactId>dubbo-security</artifactId>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <!-- dubbo -->
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-rpc-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-rpc-triple</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <!-- dubbo -->
+
+
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-protobuf</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-stub</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-netty-shaded</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java-util</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.bouncycastle</groupId>
+            <artifactId>bcprov-jdk15on</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.bouncycastle</groupId>
+            <artifactId>bcpkix-jdk15on</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.bouncycastle</groupId>
+            <artifactId>bcprov-ext-jdk15on</artifactId>
+        </dependency>
+
+        <!-- test -->
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-config-api</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <extensions>
+            <extension>
+                <groupId>kr.motd.maven</groupId>
+                <artifactId>os-maven-plugin</artifactId>
+                <version>1.6.2</version>
+            </extension>
+        </extensions>
+        <plugins>
+            <plugin>
+                <groupId>org.xolstice.maven.plugins</groupId>
+                <artifactId>protobuf-maven-plugin</artifactId>
+                <version>0.6.1</version>
+                <configuration>
+                    <protocArtifact>com.google.protobuf:protoc:${protobuf-java.version}:exe:${os.detected.classifier}</protocArtifact>
+                    <pluginId>grpc-java</pluginId>
+                    <pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}</pluginArtifact>
+                </configuration>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>compile</goal>
+                            <goal>compile-custom</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

dubbo-plugin/dubbo-security/src/main/java/org/apache/dubbo/security/cert/CertConfig.java

@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.security.cert;
+
+import java.util.Objects;
+
+import static org.apache.dubbo.security.cert.Constants.DEFAULT_REFRESH_INTERVAL;
+
+public class CertConfig {
+    private final String remoteAddress;
+    private final String envType;
+    private final String caCertPath;
+    /**
+     * Path to OpenID Connect Token file
+     */
+    private final String oidcTokenPath;
+
+    private final int refreshInterval;
+
+    public CertConfig(String remoteAddress, String envType, String caCertPath, String oidcTokenPath) {
+        this(remoteAddress, envType, caCertPath, oidcTokenPath, DEFAULT_REFRESH_INTERVAL);
+    }
+
+    public CertConfig(String remoteAddress, String envType, String caCertPath, String oidcTokenPath, int refreshInterval) {
+        this.remoteAddress = remoteAddress;
+        this.envType = envType;
+        this.caCertPath = caCertPath;
+        this.oidcTokenPath = oidcTokenPath;
+        this.refreshInterval = refreshInterval;
+    }
+
+    public String getRemoteAddress() {
+        return remoteAddress;
+    }
+
+    public String getEnvType() {
+        return envType;
+    }
+
+    public String getCaCertPath() {
+        return caCertPath;
+    }
+
+    public String getOidcTokenPath() {
+        return oidcTokenPath;
+    }
+
+    public int getRefreshInterval() {
+        return refreshInterval;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        CertConfig that = (CertConfig) o;
+        return Objects.equals(remoteAddress, that.remoteAddress) && Objects.equals(envType, that.envType) && Objects.equals(caCertPath, that.caCertPath) && Objects.equals(oidcTokenPath, that.oidcTokenPath);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(remoteAddress, envType, caCertPath, oidcTokenPath);
+    }
+}

dubbo-plugin/dubbo-security/src/main/java/org/apache/dubbo/security/cert/CertDeployerListener.java

@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.security.cert;
+
+import org.apache.dubbo.common.deploy.ApplicationDeployListener;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+
+import java.util.Objects;
+
+public class CertDeployerListener implements ApplicationDeployListener {
+    private final DubboCertManager dubboCertManager;
+
+
+    public CertDeployerListener(FrameworkModel frameworkModel) {
+        dubboCertManager = frameworkModel.getBeanFactory().getBean(DubboCertManager.class);
+    }
+
+    @Override
+    public void onStarting(ApplicationModel scopeModel) {
+        scopeModel.getApplicationConfigManager().getSsl().ifPresent(sslConfig -> {
+            if (Objects.nonNull(sslConfig.getCaAddress()) && dubboCertManager != null) {
+                CertConfig certConfig = new CertConfig(sslConfig.getCaAddress(),
+                    sslConfig.getEnvType(),
+                    sslConfig.getCaCertPath(),
+                    sslConfig.getOidcTokenPath());
+                dubboCertManager.connect(certConfig);
+            }
+        });
+    }
+
+    @Override
+    public void onStarted(ApplicationModel scopeModel) {
+    }
+
+    @Override
+    public void onStopping(ApplicationModel scopeModel) {
+        dubboCertManager.disConnect();
+    }
+
+    @Override
+    public void onStopped(ApplicationModel scopeModel) {
+
+    }
+
+    @Override
+    public void onFailure(ApplicationModel scopeModel, Throwable cause) {
+        dubboCertManager.disConnect();
+    }
+}

dubbo-plugin/dubbo-security/src/main/java/org/apache/dubbo/security/cert/CertPair.java

@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.security.cert;
+
+import java.util.Objects;
+
+public class CertPair {
+    private final String privateKey;
+    private final String certificate;
+    private final String trustCerts;
+    private final long expireTime;
+
+    public CertPair(String privateKey, String certificate, String trustCerts, long expireTime) {
+        this.privateKey = privateKey;
+        this.certificate = certificate;
+        this.trustCerts = trustCerts;
+        this.expireTime = expireTime;
+    }
+
+    public String getPrivateKey() {
+        return privateKey;
+    }
+
+    public String getCertificate() {
+        return certificate;
+    }
+
+    public String getTrustCerts() {
+        return trustCerts;
+    }
+
+    public long getExpireTime() {
+        return expireTime;
+    }
+
+    public boolean isExpire() {
+        return System.currentTimeMillis() > expireTime;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        CertPair certPair = (CertPair) o;
+        return expireTime == certPair.expireTime && Objects.equals(privateKey, certPair.privateKey) && Objects.equals(certificate, certPair.certificate) && Objects.equals(trustCerts, certPair.trustCerts);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(privateKey, certificate, trustCerts, expireTime);
+    }
+}

dubbo-plugin/dubbo-security/src/main/java/org/apache/dubbo/security/cert/CertScopeModelInitializer.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.security.cert;
+
+import org.apache.dubbo.common.beans.factory.ScopeBeanFactory;
+import org.apache.dubbo.common.utils.ClassUtils;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+import org.apache.dubbo.rpc.model.ModuleModel;
+import org.apache.dubbo.rpc.model.ScopeModelInitializer;
+
+public class CertScopeModelInitializer implements ScopeModelInitializer {
+    public static boolean isSupported() {
+        try {
+            ClassUtils.forName("io.grpc.Channel");
+            ClassUtils.forName("org.bouncycastle.pkcs.jcajce.JcaPKCS10CertificationRequestBuilder");
+            return true;
+        } catch (Throwable t) {
+            return false;
+        }
+    }
+
+    @Override
+    public void initializeFrameworkModel(FrameworkModel frameworkModel) {
+        ScopeBeanFactory beanFactory = frameworkModel.getBeanFactory();
+        if (isSupported()) {
+            beanFactory.registerBean(DubboCertManager.class);
+        }
+    }
+
+    @Override
+    public void initializeApplicationModel(ApplicationModel applicationModel) {
+
+    }
+
+    @Override
+    public void initializeModuleModel(ModuleModel moduleModel) {
+
+    }
+}

dubbo-plugin/dubbo-security/src/main/java/org/apache/dubbo/security/cert/Constants.java

@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.security.cert;
+
+public interface Constants {
+    int DEFAULT_REFRESH_INTERVAL = 30_000;
+}

dubbo-plugin/dubbo-security/src/main/java/org/apache/dubbo/security/cert/DubboCertManager.java

@@ -0,0 +1,377 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.security.cert;
+
+import org.apache.dubbo.auth.v1alpha1.DubboCertificateRequest;
+import org.apache.dubbo.auth.v1alpha1.DubboCertificateResponse;
+import org.apache.dubbo.auth.v1alpha1.DubboCertificateServiceGrpc;
+import org.apache.dubbo.common.constants.LoggerCodeConstants;
+import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
+import org.apache.dubbo.common.utils.IOUtils;
+import org.apache.dubbo.common.utils.StringUtils;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+
+import io.grpc.Channel;
+import io.grpc.Metadata;
+import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
+import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
+import io.grpc.netty.shaded.io.netty.handler.ssl.util.InsecureTrustManagerFactory;
+import org.bouncycastle.asn1.x500.X500Name;
+import org.bouncycastle.openssl.jcajce.JcaPEMWriter;
+import org.bouncycastle.operator.ContentSigner;
+import org.bouncycastle.operator.OperatorCreationException;
+import org.bouncycastle.operator.jcajce.JcaContentSignerBuilder;
+import org.bouncycastle.pkcs.PKCS10CertificationRequest;
+import org.bouncycastle.pkcs.jcajce.JcaPKCS10CertificationRequestBuilder;
+import org.bouncycastle.util.io.pem.PemObject;
+
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.StringWriter;
+import java.security.InvalidAlgorithmParameterException;
+import java.security.KeyPairGenerator;
+import java.security.NoSuchAlgorithmException;
+import java.security.PrivateKey;
+import java.security.PublicKey;
+import java.security.SecureRandom;
+import java.security.spec.ECGenParameterSpec;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+import static io.grpc.stub.MetadataUtils.newAttachHeadersInterceptor;
+import static org.apache.dubbo.common.constants.LoggerCodeConstants.CONFIG_SSL_CERT_GENERATE_FAILED;
+import static org.apache.dubbo.common.constants.LoggerCodeConstants.CONFIG_SSL_CONNECT_INSECURE;
+import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_ERROR;
+import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_FAILED_GENERATE_CERT_ISTIO;
+
+public class DubboCertManager {
+
+    private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(DubboCertManager.class);
+
+    private final FrameworkModel frameworkModel;
+    /**
+     * gRPC channel to Dubbo Cert Authority server
+     */
+    protected volatile Channel channel;
+    /**
+     * Cert pair for current Dubbo instance
+     */
+    protected volatile CertPair certPair;
+    /**
+     * Path to OpenID Connect Token file
+     */
+    protected volatile CertConfig certConfig;
+    /**
+     * Refresh cert pair for current Dubbo instance
+     */
+    protected volatile ScheduledFuture<?> refreshFuture;
+
+    public DubboCertManager(FrameworkModel frameworkModel) {
+        this.frameworkModel = frameworkModel;
+    }
+
+    public synchronized void connect(CertConfig certConfig) {
+        if (channel != null) {
+            logger.error(INTERNAL_ERROR, "", "", "Dubbo Cert Authority server is already connected.");
+            return;
+        }
+        if (certConfig == null) {
+            // No cert config, return
+            return;
+        }
+        if (StringUtils.isEmpty(certConfig.getRemoteAddress())) {
+            // No remote address configured, return
+            return;
+        }
+        if (StringUtils.isNotEmpty(certConfig.getEnvType()) && !"Kubernetes".equalsIgnoreCase(certConfig.getEnvType())) {
+            throw new IllegalArgumentException("Only support Kubernetes env now.");
+        }
+        // Create gRPC connection
+        connect0(certConfig);
+
+        this.certConfig = certConfig;
+
+        // Try to generate cert from remote
+        generateCert();
+        // Schedule refresh task
+        scheduleRefresh();
+    }
+
+    /**
+     * Create task to refresh cert pair for current Dubbo instance
+     */
+    protected void scheduleRefresh() {
+        FrameworkExecutorRepository repository = frameworkModel.getBeanFactory().getBean(FrameworkExecutorRepository.class);
+        refreshFuture = repository.getSharedScheduledExecutor().scheduleAtFixedRate(this::generateCert,
+            certConfig.getRefreshInterval(), certConfig.getRefreshInterval(), TimeUnit.MILLISECONDS);
+    }
+
+    /**
+     * Try to connect to remote certificate authorization
+     *
+     * @param certConfig certificate authorization address
+     */
+    protected void connect0(CertConfig certConfig) {
+        String caCertPath = certConfig.getCaCertPath();
+        String remoteAddress = certConfig.getRemoteAddress();
+        logger.info("Try to connect to Dubbo Cert Authority server: " + remoteAddress + ", caCertPath: " + remoteAddress);
+        try {
+            if (StringUtils.isNotEmpty(caCertPath)) {
+                channel = NettyChannelBuilder.forTarget(remoteAddress)
+                    .sslContext(
+                        GrpcSslContexts.forClient()
+                            .trustManager(new File(caCertPath))
+                            .build())
+                    .build();
+            } else {
+                logger.warn(CONFIG_SSL_CONNECT_INSECURE, "", "",
+                    "No caCertPath is provided, will use insecure connection.");
+                channel = NettyChannelBuilder.forTarget(remoteAddress)
+                    .sslContext(GrpcSslContexts.forClient()
+                        .trustManager(InsecureTrustManagerFactory.INSTANCE)
+                        .build())
+                    .build();
+            }
+        } catch (Exception e) {
+            logger.error(LoggerCodeConstants.CONFIG_SSL_PATH_LOAD_FAILED, "", "", "Failed to load SSL cert file.", e);
+            throw new RuntimeException(e);
+        }
+    }
+
+    public synchronized void disConnect() {
+        if (refreshFuture != null) {
+            refreshFuture.cancel(true);
+            refreshFuture = null;
+        }
+        if (channel != null) {
+            channel = null;
+        }
+    }
+
+    public boolean isConnected() {
+        return certConfig != null && channel != null && certPair != null;
+    }
+
+    protected CertPair generateCert() {
+        if (certPair != null && !certPair.isExpire()) {
+            return certPair;
+        }
+        synchronized (this) {
+            if (certPair == null || certPair.isExpire()) {
+                try {
+                    logger.info("Try to generate cert from Dubbo Certificate Authority.");
+                    CertPair certFromRemote = refreshCert();
+                    if (certFromRemote != null) {
+                        certPair = certFromRemote;
+                    } else {
+                        logger.error(CONFIG_SSL_CERT_GENERATE_FAILED, "", "", "Generate Cert from Dubbo Certificate Authority failed.");
+                    }
+                } catch (Exception e) {
+                    logger.error(REGISTRY_FAILED_GENERATE_CERT_ISTIO, "", "", "Generate Cert from Istio failed.", e);
+                }
+            }
+        }
+        return certPair;
+    }
+
+    /**
+     * Request remote certificate authorization to generate cert pair for current Dubbo instance
+     *
+     * @return cert pair
+     * @throws IOException ioException
+     */
+    protected CertPair refreshCert() throws IOException {
+        KeyPair keyPair = signWithEcdsa();
+
+        if (keyPair == null) {
+            keyPair = signWithRsa();
+        }
+
+        if (keyPair == null) {
+            logger.error(CONFIG_SSL_CERT_GENERATE_FAILED, "", "", "Generate Key failed. Please check if your system support.");
+            return null;
+        }
+
+        String csr = generateCsr(keyPair);
+        DubboCertificateServiceGrpc.DubboCertificateServiceBlockingStub stub = DubboCertificateServiceGrpc.newBlockingStub(channel);
+        stub = setHeaderIfNeed(stub);
+
+        String privateKeyPem = generatePrivatePemKey(keyPair);
+        DubboCertificateResponse certificateResponse = stub.createCertificate(generateRequest(csr));
+
+        if (certificateResponse == null || !certificateResponse.getSuccess()) {
+            logger.error(CONFIG_SSL_CERT_GENERATE_FAILED, "", "", "Failed to generate cert from Dubbo Certificate Authority. " +
+                "Message: " + (certificateResponse == null ? "null" : certificateResponse.getMessage()));
+            return null;
+        }
+        logger.info("Successfully generate cert from Dubbo Certificate Authority. Cert expire time: " + certificateResponse.getExpireTime());
+
+        return new CertPair(privateKeyPem,
+            certificateResponse.getCertPem(),
+            String.join("\n", certificateResponse.getTrustCertsList()),
+            certificateResponse.getExpireTime());
+    }
+
+    private DubboCertificateServiceGrpc.DubboCertificateServiceBlockingStub setHeaderIfNeed(DubboCertificateServiceGrpc.DubboCertificateServiceBlockingStub stub) throws IOException {
+        String oidcTokenPath = certConfig.getOidcTokenPath();
+        if (StringUtils.isNotEmpty(oidcTokenPath)) {
+            Metadata header = new Metadata();
+            Metadata.Key<String> key = Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER);
+            header.put(key, "Bearer " +
+                IOUtils.read(new FileReader(oidcTokenPath))
+                    .replace("\n", "")
+                    .replace("\t", "")
+                    .replace("\r", "")
+                    .trim());
+
+            stub = stub.withInterceptors(newAttachHeadersInterceptor(header));
+            logger.info("Use oidc token from " + oidcTokenPath + " to connect to Dubbo Certificate Authority.");
+        } else {
+            logger.warn(CONFIG_SSL_CONNECT_INSECURE, "", "",
+                "Use insecure connection to connect to Dubbo Certificate Authority. Reason: No oidc token is provided.");
+        }
+        return stub;
+    }
+
+    /**
+     * Generate key pair with RSA
+     *
+     * @return key pair
+     */
+    protected static KeyPair signWithRsa() {
+        KeyPair keyPair = null;
+        try {
+            KeyPairGenerator kpGenerator = KeyPairGenerator.getInstance("RSA");
+            kpGenerator.initialize(4096);
+            java.security.KeyPair keypair = kpGenerator.generateKeyPair();
+            PublicKey publicKey = keypair.getPublic();
+            PrivateKey privateKey = keypair.getPrivate();
+            ContentSigner signer = new JcaContentSignerBuilder("SHA256WithRSA").build(keypair.getPrivate());
+            keyPair = new KeyPair(publicKey, privateKey, signer);
+        } catch (NoSuchAlgorithmException | OperatorCreationException e) {
+            logger.error(CONFIG_SSL_CERT_GENERATE_FAILED, "", "", "Generate Key with SHA256WithRSA algorithm failed. Please check if your system support.", e);
+        }
+        return keyPair;
+    }
+
+    /**
+     * Generate key pair with ECDSA
+     *
+     * @return key pair
+     */
+    protected static KeyPair signWithEcdsa() {
+        KeyPair keyPair = null;
+        try {
+            ECGenParameterSpec ecSpec = new ECGenParameterSpec("secp256r1");
+            KeyPairGenerator g = KeyPairGenerator.getInstance("EC");
+            g.initialize(ecSpec, new SecureRandom());
+            java.security.KeyPair keypair = g.generateKeyPair();
+            PublicKey publicKey = keypair.getPublic();
+            PrivateKey privateKey = keypair.getPrivate();
+            ContentSigner signer = new JcaContentSignerBuilder("SHA256withECDSA").build(privateKey);
+            keyPair = new KeyPair(publicKey, privateKey, signer);
+        } catch (NoSuchAlgorithmException | InvalidAlgorithmParameterException | OperatorCreationException e) {
+            logger.error(CONFIG_SSL_CERT_GENERATE_FAILED, "", "", "Generate Key with secp256r1 algorithm failed. Please check if your system support. "
+                + "Will attempt to generate with RSA2048.", e);
+        }
+        return keyPair;
+    }
+
+    private DubboCertificateRequest generateRequest(String csr) {
+        return DubboCertificateRequest.newBuilder().setCsr(csr).setType("CONNECTION").build();
+    }
+
+    /**
+     * Generate private key in pem encoded
+     *
+     * @param keyPair key pair
+     * @return private key
+     * @throws IOException ioException
+     */
+    private String generatePrivatePemKey(KeyPair keyPair) throws IOException {
+        String key = generatePemKey("RSA PRIVATE KEY", keyPair.getPrivateKey().getEncoded());
+        if (logger.isDebugEnabled()) {
+            logger.debug("Generated Private Key. \n" + key);
+        }
+        return key;
+    }
+
+    /**
+     * Generate content in pem encoded
+     *
+     * @param type    content type
+     * @param content content
+     * @return encoded data
+     * @throws IOException ioException
+     */
+    private String generatePemKey(String type, byte[] content) throws IOException {
+        PemObject pemObject = new PemObject(type, content);
+        StringWriter str = new StringWriter();
+        JcaPEMWriter jcaPEMWriter = new JcaPEMWriter(str);
+        jcaPEMWriter.writeObject(pemObject);
+        jcaPEMWriter.close();
+        str.close();
+        return str.toString();
+    }
+
+    /**
+     * Generate CSR (Certificate Sign Request)
+     *
+     * @param keyPair key pair to request
+     * @return csr
+     * @throws IOException ioException
+     */
+    private String generateCsr(KeyPair keyPair) throws IOException {
+        PKCS10CertificationRequest request = new JcaPKCS10CertificationRequestBuilder(
+            new X500Name("O=" + "cluster.domain"), keyPair.getPublicKey())
+            .build(keyPair.getSigner());
+
+        String csr = generatePemKey("CERTIFICATE REQUEST", request.getEncoded());
+
+        if (logger.isDebugEnabled()) {
+            logger.debug("CSR Request to Dubbo Certificate Authorization. \n" + csr);
+        }
+        return csr;
+    }
+
+    protected static class KeyPair {
+        private final PublicKey publicKey;
+        private final PrivateKey privateKey;
+        private final ContentSigner signer;
+
+        public KeyPair(PublicKey publicKey, PrivateKey privateKey, ContentSigner signer) {
+            this.publicKey = publicKey;
+            this.privateKey = privateKey;
+            this.signer = signer;
+        }
+
+        public PublicKey getPublicKey() {
+            return publicKey;
+        }
+
+        public PrivateKey getPrivateKey() {
+            return privateKey;
+        }
+
+        public ContentSigner getSigner() {
+            return signer;
+        }
+    }
+}

dubbo-plugin/dubbo-security/src/main/java/org/apache/dubbo/security/cert/DubboCertProvider.java

@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.security.cert;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.Activate;
+import org.apache.dubbo.common.ssl.AuthPolicy;
+import org.apache.dubbo.common.ssl.Cert;
+import org.apache.dubbo.common.ssl.CertProvider;
+import org.apache.dubbo.common.ssl.ProviderCert;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+
+import java.nio.charset.StandardCharsets;
+
+@Activate
+public class DubboCertProvider implements CertProvider {
+    private final DubboCertManager dubboCertManager;
+
+
+    public DubboCertProvider(FrameworkModel frameworkModel) {
+        dubboCertManager = frameworkModel.getBeanFactory().getBean(DubboCertManager.class);
+    }
+
+    @Override
+    public boolean isSupport(URL address) {
+        return dubboCertManager != null && dubboCertManager.isConnected();
+    }
+
+    @Override
+    public ProviderCert getProviderConnectionConfig(URL localAddress) {
+        CertPair certPair = dubboCertManager.generateCert();
+        if (certPair == null) {
+            return null;
+        }
+        return new ProviderCert(certPair.getCertificate().getBytes(StandardCharsets.UTF_8),
+            certPair.getPrivateKey().getBytes(StandardCharsets.UTF_8),
+            certPair.getTrustCerts().getBytes(StandardCharsets.UTF_8), AuthPolicy.NONE);
+    }
+
+    @Override
+    public Cert getConsumerConnectionConfig(URL remoteAddress) {
+        CertPair certPair = dubboCertManager.generateCert();
+        if (certPair == null) {
+            return null;
+        }
+        return new Cert(certPair.getCertificate().getBytes(StandardCharsets.UTF_8),
+            certPair.getPrivateKey().getBytes(StandardCharsets.UTF_8),
+            certPair.getTrustCerts().getBytes(StandardCharsets.UTF_8));
+    }
+}

dubbo-plugin/dubbo-security/src/main/proto/ca.proto

@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+syntax = "proto3";
+
+import "google/protobuf/struct.proto";
+
+package org.apache.dubbo.auth.v1alpha1;
+
+option go_package = "github.com/apache/dubbo-admin/ca/v1alpha1";
+option java_multiple_files = true;
+
+
+message DubboCertificateRequest {
+  string csr = 1;
+  string type = 2;
+
+  google.protobuf.Struct metadata = 3;
+}
+
+message DubboCertificateResponse {
+  bool success = 1;
+  string cert_pem = 2;
+  repeated string trust_certs = 3;
+  int64 expire_time = 4;
+  string message = 5;
+}
+
+service DubboCertificateService {
+  rpc CreateCertificate(DubboCertificateRequest)
+      returns (DubboCertificateResponse) {
+  }
+}

dubbo-plugin/dubbo-security/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.common.deploy.ApplicationDeployListener

@@ -0,0 +1 @@
+cert=org.apache.dubbo.security.cert.CertDeployerListener

dubbo-plugin/dubbo-security/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.common.ssl.CertProvider

@@ -0,0 +1 @@
+dubbo=org.apache.dubbo.security.cert.DubboCertProvider

dubbo-plugin/dubbo-security/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.model.ScopeModelInitializer

@@ -0,0 +1 @@
+cert=org.apache.dubbo.security.cert.CertScopeModelInitializer

dubbo-plugin/dubbo-security/src/test/java/org/apache/dubbo/security/cert/CertDeployerListenerTest.java

@@ -0,0 +1,242 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.security.cert;
+
+import org.apache.dubbo.common.deploy.ApplicationDeployer;
+import org.apache.dubbo.config.ApplicationConfig;
+import org.apache.dubbo.config.MetadataReportConfig;
+import org.apache.dubbo.config.SslConfig;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+import org.mockito.Mockito;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+class CertDeployerListenerTest {
+    @Test
+    void testEmpty1() {
+        AtomicReference<DubboCertManager> reference = new AtomicReference<>();
+        try (MockedConstruction<DubboCertManager> construction =
+                 Mockito.mockConstruction(DubboCertManager.class, (mock, context) -> {
+                     reference.set(mock);
+                 })) {
+            FrameworkModel frameworkModel = new FrameworkModel();
+            ApplicationModel applicationModel = frameworkModel.newApplication();
+            applicationModel.getApplicationConfigManager().setApplication(new ApplicationConfig("test"));
+            applicationModel.getDeployer().start();
+            Mockito.verify(reference.get(), Mockito.times(0))
+                .connect(Mockito.any());
+            applicationModel.getDeployer().stop();
+            Mockito.verify(reference.get(), Mockito.atLeast(1)).disConnect();
+            frameworkModel.destroy();
+        }
+    }
+
+    @Test
+    void testEmpty2() {
+        AtomicReference<DubboCertManager> reference = new AtomicReference<>();
+        try (MockedConstruction<DubboCertManager> construction =
+                 Mockito.mockConstruction(DubboCertManager.class, (mock, context) -> {
+                     reference.set(mock);
+                 })) {
+            FrameworkModel frameworkModel = new FrameworkModel();
+            ApplicationModel applicationModel = frameworkModel.newApplication();
+            applicationModel.getApplicationConfigManager().setApplication(new ApplicationConfig("test"));
+            applicationModel.getApplicationConfigManager().setSsl(new SslConfig());
+            applicationModel.getDeployer().start();
+            Mockito.verify(reference.get(), Mockito.times(0))
+                .connect(Mockito.any());
+            applicationModel.getDeployer().stop();
+            Mockito.verify(reference.get(), Mockito.atLeast(1)).disConnect();
+            frameworkModel.destroy();
+        }
+    }
+
+    @Test
+    void testCreate() {
+        AtomicReference<DubboCertManager> reference = new AtomicReference<>();
+        try (MockedConstruction<DubboCertManager> construction =
+                 Mockito.mockConstruction(DubboCertManager.class, (mock, context) -> {
+                     reference.set(mock);
+                 })) {
+            FrameworkModel frameworkModel = new FrameworkModel();
+            ApplicationModel applicationModel = frameworkModel.newApplication();
+            applicationModel.getApplicationConfigManager().setApplication(new ApplicationConfig("test"));
+            SslConfig sslConfig = new SslConfig();
+            sslConfig.setCaAddress("127.0.0.1:30060");
+            applicationModel.getApplicationConfigManager().setSsl(sslConfig);
+
+            applicationModel.getDeployer().start();
+            Mockito.verify(reference.get(), Mockito.times(1))
+                .connect(Mockito.any());
+            applicationModel.getDeployer().stop();
+            Mockito.verify(reference.get(), Mockito.atLeast(1))
+                .disConnect();
+            frameworkModel.destroy();
+        }
+    }
+
+    @Test
+    void testFailure() {
+        AtomicReference<DubboCertManager> reference = new AtomicReference<>();
+        try (MockedConstruction<DubboCertManager> construction =
+                 Mockito.mockConstruction(DubboCertManager.class, (mock, context) -> {
+                     reference.set(mock);
+                 })) {
+            FrameworkModel frameworkModel = new FrameworkModel();
+            ApplicationModel applicationModel = frameworkModel.newApplication();
+            applicationModel.getApplicationConfigManager().setApplication(new ApplicationConfig("test"));
+            SslConfig sslConfig = new SslConfig();
+            sslConfig.setCaAddress("127.0.0.1:30060");
+            applicationModel.getApplicationConfigManager().setSsl(sslConfig);
+            applicationModel.getApplicationConfigManager().addMetadataReport(new MetadataReportConfig("absent"));
+
+            ApplicationDeployer deployer = applicationModel.getDeployer();
+            Assertions.assertThrows(IllegalArgumentException.class, deployer::start);
+            Mockito.verify(reference.get(), Mockito.times(1))
+                .connect(Mockito.any());
+            Mockito.verify(reference.get(), Mockito.atLeast(1))
+                .disConnect();
+            frameworkModel.destroy();
+        }
+    }
+
+    @Test
+    void testNotFound1() {
+        ClassLoader originClassLoader = Thread.currentThread().getContextClassLoader();
+        ClassLoader newClassLoader = new ClassLoader(originClassLoader) {
+            @Override
+            public Class<?> loadClass(String name) throws ClassNotFoundException {
+                if (name.startsWith("io.grpc.Channel")) {
+                    throw new ClassNotFoundException("Test");
+                }
+                return super.loadClass(name);
+            }
+        };
+        Thread.currentThread().setContextClassLoader(newClassLoader);
+        try (MockedConstruction<DubboCertManager> construction =
+                 Mockito.mockConstruction(DubboCertManager.class, (mock, context) -> {
+                     // ignore
+                 })) {
+            FrameworkModel frameworkModel = new FrameworkModel();
+            ApplicationModel applicationModel = frameworkModel.newApplication();
+            applicationModel.getApplicationConfigManager().setApplication(new ApplicationConfig("test"));
+            SslConfig sslConfig = new SslConfig();
+            sslConfig.setCaAddress("127.0.0.1:30060");
+            applicationModel.getApplicationConfigManager().setSsl(sslConfig);
+
+            applicationModel.getDeployer().start();
+            applicationModel.getDeployer().stop();
+            Assertions.assertEquals(0, construction.constructed().size());
+            frameworkModel.destroy();
+        }
+        Thread.currentThread().setContextClassLoader(originClassLoader);
+    }
+
+    @Test
+    void testNotFound2() {
+        ClassLoader originClassLoader = Thread.currentThread().getContextClassLoader();
+        ClassLoader newClassLoader = new ClassLoader(originClassLoader) {
+            @Override
+            public Class<?> loadClass(String name) throws ClassNotFoundException {
+                if (name.startsWith("org.bouncycastle.pkcs.jcajce.JcaPKCS10CertificationRequestBuilder")) {
+                    throw new ClassNotFoundException("Test");
+                }
+                return super.loadClass(name);
+            }
+        };
+        Thread.currentThread().setContextClassLoader(newClassLoader);
+        try (MockedConstruction<DubboCertManager> construction =
+                 Mockito.mockConstruction(DubboCertManager.class, (mock, context) -> {
+                     // ignore
+                 })) {
+            FrameworkModel frameworkModel = new FrameworkModel();
+            ApplicationModel applicationModel = frameworkModel.newApplication();
+            applicationModel.getApplicationConfigManager().setApplication(new ApplicationConfig("test"));
+            SslConfig sslConfig = new SslConfig();
+            sslConfig.setCaAddress("127.0.0.1:30060");
+            applicationModel.getApplicationConfigManager().setSsl(sslConfig);
+
+            applicationModel.getDeployer().start();
+            applicationModel.getDeployer().stop();
+            Assertions.assertEquals(0, construction.constructed().size());
+            frameworkModel.destroy();
+        }
+        Thread.currentThread().setContextClassLoader(originClassLoader);
+    }
+
+    @Test
+    void testParams1() {
+        AtomicReference<DubboCertManager> reference = new AtomicReference<>();
+        try (MockedConstruction<DubboCertManager> construction =
+                 Mockito.mockConstruction(DubboCertManager.class, (mock, context) -> {
+                     reference.set(mock);
+                 })) {
+            FrameworkModel frameworkModel = new FrameworkModel();
+            ApplicationModel applicationModel = frameworkModel.newApplication();
+            applicationModel.getApplicationConfigManager().setApplication(new ApplicationConfig("test"));
+            SslConfig sslConfig = new SslConfig();
+            sslConfig.setCaAddress("127.0.0.1:30060");
+            sslConfig.setCaCertPath("certs/ca.crt");
+            sslConfig.setOidcTokenPath("token");
+            sslConfig.setEnvType("test");
+            applicationModel.getApplicationConfigManager().setSsl(sslConfig);
+
+            applicationModel.getDeployer().start();
+            Mockito.verify(reference.get(), Mockito.times(1))
+                .connect(new CertConfig("127.0.0.1:30060", "test", "certs/ca.crt", "token"));
+            applicationModel.getDeployer().stop();
+            Mockito.verify(reference.get(), Mockito.atLeast(1))
+                .disConnect();
+            frameworkModel.destroy();
+        }
+    }
+
+    @Disabled("Enable me until properties from envs work.")
+    @Test
+    void testParams2() {
+        AtomicReference<DubboCertManager> reference = new AtomicReference<>();
+        try (MockedConstruction<DubboCertManager> construction =
+                 Mockito.mockConstruction(DubboCertManager.class, (mock, context) -> {
+                     reference.set(mock);
+                 })) {
+            System.setProperty("dubbo.ssl.ca-address", "127.0.0.1:30060");
+            System.setProperty("dubbo.ssl.ca-cert-path", "certs/ca.crt");
+            System.setProperty("dubbo.ssl.oidc-token-path", "token");
+            System.setProperty("dubbo.ssl.env-type", "test");
+            FrameworkModel frameworkModel = new FrameworkModel();
+            ApplicationModel applicationModel = frameworkModel.newApplication();
+            applicationModel.getApplicationConfigManager().setApplication(new ApplicationConfig("test"));
+
+            applicationModel.getDeployer().start();
+            Mockito.verify(reference.get(), Mockito.times(1))
+                .connect(new CertConfig("127.0.0.1:30060", "test", "certs/ca.crt", "token"));
+            applicationModel.getDeployer().stop();
+            Mockito.verify(reference.get(), Mockito.atLeast(1)).disConnect();
+            frameworkModel.destroy();
+            System.clearProperty("dubbo.ssl.ca-address");
+            System.clearProperty("dubbo.ssl.ca-cert-path");
+            System.clearProperty("dubbo.ssl.oidc-token-path");
+            System.clearProperty("dubbo.ssl.env-type");
+        }
+    }
+}

dubbo-plugin/dubbo-security/src/test/java/org/apache/dubbo/security/cert/DubboCertManagerTest.java

@@ -0,0 +1,284 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.security.cert;
+
+import org.apache.dubbo.auth.v1alpha1.DubboCertificateResponse;
+import org.apache.dubbo.auth.v1alpha1.DubboCertificateServiceGrpc;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+
+import io.grpc.Channel;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.awaitility.Awaitility.await;
+import static org.mockito.Answers.CALLS_REAL_METHODS;
+
+class DubboCertManagerTest {
+    @Test
+    void test1() {
+        FrameworkModel frameworkModel = new FrameworkModel();
+        DubboCertManager certManager = new DubboCertManager(frameworkModel) {
+            @Override
+            protected void connect0(CertConfig certConfig) {
+                Assertions.assertEquals("127.0.0.1:30060", certConfig.getRemoteAddress());
+                Assertions.assertEquals("caCertPath", certConfig.getCaCertPath());
+            }
+
+            @Override
+            protected CertPair generateCert() {
+                return null;
+            }
+
+            @Override
+            protected void scheduleRefresh() {
+
+            }
+        };
+        certManager.connect(new CertConfig("127.0.0.1:30060", null, "caCertPath", "oidc"));
+        Assertions.assertEquals(new CertConfig("127.0.0.1:30060", null, "caCertPath", "oidc"), certManager.certConfig);
+
+        certManager.connect(new CertConfig("127.0.0.1:30060", "Kubernetes", "caCertPath", "oidc123"));
+        Assertions.assertEquals(new CertConfig("127.0.0.1:30060", "Kubernetes", "caCertPath", "oidc123"), certManager.certConfig);
+
+        certManager.connect(new CertConfig("127.0.0.1:30060", "kubernetes", "caCertPath", "oidc345"));
+        Assertions.assertEquals(new CertConfig("127.0.0.1:30060", "kubernetes", "caCertPath", "oidc345"), certManager.certConfig);
+
+        CertConfig certConfig = new CertConfig("127.0.0.1:30060", "vm", "caCertPath", "oidc");
+        Assertions.assertThrows(IllegalArgumentException.class, () -> certManager.connect(certConfig));
+        Assertions.assertEquals(new CertConfig("127.0.0.1:30060", "kubernetes", "caCertPath", "oidc345"), certManager.certConfig);
+
+        certManager.connect(null);
+        Assertions.assertEquals(new CertConfig("127.0.0.1:30060", "kubernetes", "caCertPath", "oidc345"), certManager.certConfig);
+
+        certManager.connect(new CertConfig(null, null, null, null));
+        Assertions.assertEquals(new CertConfig("127.0.0.1:30060", "kubernetes", "caCertPath", "oidc345"), certManager.certConfig);
+
+        certManager.channel = Mockito.mock(Channel.class);
+        certManager.connect(new CertConfig("error", null, "error", "error"));
+        Assertions.assertEquals(new CertConfig("127.0.0.1:30060", "kubernetes", "caCertPath", "oidc345"), certManager.certConfig);
+
+        frameworkModel.destroy();
+    }
+
+    @Test
+    void testRefresh() {
+        FrameworkModel frameworkModel = new FrameworkModel();
+        AtomicInteger count = new AtomicInteger(0);
+        DubboCertManager certManager = new DubboCertManager(frameworkModel) {
+            @Override
+            protected CertPair generateCert() {
+                count.incrementAndGet();
+                return null;
+            }
+        };
+
+        certManager.certConfig = new CertConfig(null, null, null, null, 10);
+        certManager.scheduleRefresh();
+
+        Assertions.assertNotNull(certManager.refreshFuture);
+        await().until(() -> count.get() > 1);
+        certManager.refreshFuture.cancel(false);
+        frameworkModel.destroy();
+    }
+
+    @Test
+    void testConnect1() {
+        FrameworkModel frameworkModel = new FrameworkModel();
+        DubboCertManager certManager = new DubboCertManager(frameworkModel);
+        CertConfig certConfig = new CertConfig("127.0.0.1:30062", null, null, null);
+        certManager.connect0(certConfig);
+        Assertions.assertNotNull(certManager.channel);
+        Assertions.assertEquals("127.0.0.1:30062", certManager.channel.authority());
+
+        frameworkModel.destroy();
+    }
+
+    @Test
+    void testConnect2() {
+        FrameworkModel frameworkModel = new FrameworkModel();
+        DubboCertManager certManager = new DubboCertManager(frameworkModel);
+        String file = this.getClass().getClassLoader().getResource("certs/ca.crt").getFile();
+        CertConfig certConfig = new CertConfig("127.0.0.1:30062", null, file, null);
+        certManager.connect0(certConfig);
+        Assertions.assertNotNull(certManager.channel);
+        Assertions.assertEquals("127.0.0.1:30062", certManager.channel.authority());
+
+        frameworkModel.destroy();
+    }
+
+    @Test
+    void testConnect3() {
+        FrameworkModel frameworkModel = new FrameworkModel();
+        DubboCertManager certManager = new DubboCertManager(frameworkModel);
+        String file = this.getClass().getClassLoader().getResource("certs/broken-ca.crt").getFile();
+        CertConfig certConfig = new CertConfig("127.0.0.1:30062", null, file, null);
+        Assertions.assertThrows(RuntimeException.class, () -> certManager.connect0(certConfig));
+
+        frameworkModel.destroy();
+    }
+
+    @Test
+    void testDisconnect() {
+        FrameworkModel frameworkModel = new FrameworkModel();
+        DubboCertManager certManager = new DubboCertManager(frameworkModel);
+        ScheduledFuture scheduledFuture = Mockito.mock(ScheduledFuture.class);
+        certManager.refreshFuture = scheduledFuture;
+        certManager.disConnect();
+        Assertions.assertNull(certManager.refreshFuture);
+        Mockito.verify(scheduledFuture, Mockito.times(1)).cancel(true);
+
+
+        certManager.channel = Mockito.mock(Channel.class);
+        certManager.disConnect();
+        Assertions.assertNull(certManager.channel);
+
+        frameworkModel.destroy();
+    }
+
+    @Test
+    void testConnected() {
+        FrameworkModel frameworkModel = new FrameworkModel();
+        DubboCertManager certManager = new DubboCertManager(frameworkModel);
+
+        Assertions.assertFalse(certManager.isConnected());
+
+        certManager.certConfig = Mockito.mock(CertConfig.class);
+        Assertions.assertFalse(certManager.isConnected());
+
+        certManager.channel = Mockito.mock(Channel.class);
+        Assertions.assertFalse(certManager.isConnected());
+
+        certManager.certPair = Mockito.mock(CertPair.class);
+        Assertions.assertTrue(certManager.isConnected());
+
+        frameworkModel.destroy();
+    }
+
+    @Test
+    void testGenerateCert() {
+        FrameworkModel frameworkModel = new FrameworkModel();
+
+        AtomicBoolean exception = new AtomicBoolean(false);
+        AtomicReference<CertPair> certPairReference = new AtomicReference<>();
+        DubboCertManager certManager = new DubboCertManager(frameworkModel) {
+            @Override
+            protected CertPair refreshCert() throws IOException {
+                if (exception.get()) {
+                    throw new IOException("test");
+                }
+                return certPairReference.get();
+            }
+        };
+
+        CertPair certPair = new CertPair("", "", "", Long.MAX_VALUE);
+        certPairReference.set(certPair);
+
+        Assertions.assertEquals(certPair, certManager.generateCert());
+
+        certManager.certPair = new CertPair("", "", "", Long.MAX_VALUE - 10000);
+        Assertions.assertEquals(new CertPair("", "", "", Long.MAX_VALUE - 10000), certManager.generateCert());
+
+        certManager.certPair = new CertPair("", "", "", 0);
+        Assertions.assertEquals(certPair, certManager.generateCert());
+
+        certManager.certPair = new CertPair("", "", "", 0);
+        certPairReference.set(null);
+        Assertions.assertEquals(new CertPair("", "", "", 0), certManager.generateCert());
+
+        exception.set(true);
+        Assertions.assertEquals(new CertPair("", "", "", 0), certManager.generateCert());
+
+        frameworkModel.destroy();
+    }
+
+    @Test
+    void testSignWithRsa() {
+        DubboCertManager.KeyPair keyPair = DubboCertManager.signWithRsa();
+        Assertions.assertNotNull(keyPair);
+        Assertions.assertNotNull(keyPair.getPrivateKey());
+        Assertions.assertNotNull(keyPair.getPublicKey());
+        Assertions.assertNotNull(keyPair.getSigner());
+    }
+
+    @Test
+    void testSignWithEcdsa() {
+        DubboCertManager.KeyPair keyPair = DubboCertManager.signWithEcdsa();
+        Assertions.assertNotNull(keyPair);
+        Assertions.assertNotNull(keyPair.getPrivateKey());
+        Assertions.assertNotNull(keyPair.getPublicKey());
+        Assertions.assertNotNull(keyPair.getSigner());
+    }
+
+
+    @Test
+    void testRefreshCert() throws IOException {
+        try (MockedStatic<DubboCertManager> managerMock = Mockito.mockStatic(DubboCertManager.class, CALLS_REAL_METHODS)) {
+            FrameworkModel frameworkModel = new FrameworkModel();
+            DubboCertManager certManager = new DubboCertManager(frameworkModel);
+            managerMock.when(DubboCertManager::signWithEcdsa).thenReturn(null);
+            managerMock.when(DubboCertManager::signWithRsa).thenReturn(null);
+
+            Assertions.assertNull(certManager.refreshCert());
+
+            managerMock.when(DubboCertManager::signWithEcdsa).thenCallRealMethod();
+
+            certManager.channel = Mockito.mock(Channel.class);
+            try (MockedStatic<DubboCertificateServiceGrpc> mockGrpc = Mockito.mockStatic(DubboCertificateServiceGrpc.class, CALLS_REAL_METHODS)) {
+                DubboCertificateServiceGrpc.DubboCertificateServiceBlockingStub stub = Mockito.mock(DubboCertificateServiceGrpc.DubboCertificateServiceBlockingStub.class);
+                mockGrpc.when(() -> DubboCertificateServiceGrpc.newBlockingStub(Mockito.any(Channel.class))).thenReturn(stub);
+                Mockito.when(stub.createCertificate(Mockito.any())).thenReturn(
+                    DubboCertificateResponse.newBuilder()
+                    .setSuccess(false).build());
+
+                certManager.certConfig = new CertConfig(null, null, null, null);
+                Assertions.assertNull(certManager.refreshCert());
+
+                String file = this.getClass().getClassLoader().getResource("certs/token").getFile();
+                Mockito.when(stub.withInterceptors(Mockito.any())).thenReturn(stub);
+                certManager.certConfig = new CertConfig(null, null, null, file);
+
+                Assertions.assertNull(certManager.refreshCert());
+                Mockito.verify(stub, Mockito.times(1)).withInterceptors(Mockito.any());
+
+                Mockito.when(stub.createCertificate(Mockito.any())).thenReturn(
+                    DubboCertificateResponse.newBuilder()
+                        .setSuccess(true)
+                        .setCertPem("certPem")
+                        .addTrustCerts("trustCerts")
+                        .setExpireTime(123456).build());
+                CertPair certPair = certManager.refreshCert();
+                Assertions.assertNotNull(certPair);
+                Assertions.assertEquals("certPem", certPair.getCertificate());
+                Assertions.assertEquals("trustCerts", certPair.getTrustCerts());
+                Assertions.assertEquals(123456, certPair.getExpireTime());
+
+                Mockito.when(stub.createCertificate(Mockito.any())).thenReturn(null);
+                Assertions.assertNull(certManager.refreshCert());
+            }
+
+            frameworkModel.destroy();
+        }
+    }
+}

dubbo-plugin/dubbo-security/src/test/java/org/apache/dubbo/security/cert/DubboCertProviderTest.java

@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.security.cert;
+
+import org.apache.dubbo.common.ssl.AuthPolicy;
+import org.apache.dubbo.common.ssl.Cert;
+import org.apache.dubbo.common.ssl.ProviderCert;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+import org.mockito.Mockito;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+class DubboCertProviderTest {
+    @Test
+    void testEnable() {
+        AtomicReference<DubboCertManager> reference = new AtomicReference<>();
+        try (MockedConstruction<DubboCertManager> construction =
+                 Mockito.mockConstruction(DubboCertManager.class, (mock, context) -> {
+                     reference.set(mock);
+                 })) {
+            FrameworkModel frameworkModel = new FrameworkModel();
+            DubboCertProvider provider = new DubboCertProvider(frameworkModel);
+
+            Mockito.when(reference.get().isConnected()).thenReturn(true);
+            Assertions.assertTrue(provider.isSupport(null));
+
+            Mockito.when(reference.get().isConnected()).thenReturn(false);
+            Assertions.assertFalse(provider.isSupport(null));
+
+            frameworkModel.destroy();
+        }
+    }
+
+    @Test
+    void testEnable1() {
+        ClassLoader originClassLoader = Thread.currentThread().getContextClassLoader();
+        ClassLoader newClassLoader = new ClassLoader(originClassLoader) {
+            @Override
+            public Class<?> loadClass(String name) throws ClassNotFoundException {
+                if (name.startsWith("io.grpc.Channel")) {
+                    throw new ClassNotFoundException("Test");
+                }
+                return super.loadClass(name);
+            }
+        };
+        Thread.currentThread().setContextClassLoader(newClassLoader);
+        try (MockedConstruction<DubboCertManager> construction =
+                 Mockito.mockConstruction(DubboCertManager.class, (mock, context) -> {
+                     // ignore
+                 })) {
+            FrameworkModel frameworkModel = new FrameworkModel();
+            DubboCertProvider provider = new DubboCertProvider(frameworkModel);
+
+            Assertions.assertFalse(provider.isSupport(null));
+
+            frameworkModel.destroy();
+        }
+        Thread.currentThread().setContextClassLoader(originClassLoader);
+    }
+
+    @Test
+    void testEnable2() {
+        ClassLoader originClassLoader = Thread.currentThread().getContextClassLoader();
+        ClassLoader newClassLoader = new ClassLoader(originClassLoader) {
+            @Override
+            public Class<?> loadClass(String name) throws ClassNotFoundException {
+                if (name.startsWith("org.bouncycastle.pkcs.jcajce.JcaPKCS10CertificationRequestBuilder")) {
+                    throw new ClassNotFoundException("Test");
+                }
+                return super.loadClass(name);
+            }
+        };
+        Thread.currentThread().setContextClassLoader(newClassLoader);
+        try (MockedConstruction<DubboCertManager> construction =
+                 Mockito.mockConstruction(DubboCertManager.class, (mock, context) -> {
+                     // ignore
+                 })) {
+            FrameworkModel frameworkModel = new FrameworkModel();
+            DubboCertProvider provider = new DubboCertProvider(frameworkModel);
+
+            Assertions.assertFalse(provider.isSupport(null));
+
+            frameworkModel.destroy();
+        }
+        Thread.currentThread().setContextClassLoader(originClassLoader);
+    }
+
+    @Test
+    void getProviderConnectionConfigTest() {
+        AtomicReference<DubboCertManager> reference = new AtomicReference<>();
+        try (MockedConstruction<DubboCertManager> construction =
+                 Mockito.mockConstruction(DubboCertManager.class, (mock, context) -> {
+                     reference.set(mock);
+                 })) {
+            FrameworkModel frameworkModel = new FrameworkModel();
+            DubboCertProvider provider = new DubboCertProvider(frameworkModel);
+            Assertions.assertNull(provider.getProviderConnectionConfig(null));
+
+            CertPair certPair = new CertPair("privateKey", "publicKey", "trustCerts", 12345);
+            Mockito.when(reference.get().generateCert()).thenReturn(certPair);
+            ProviderCert providerConnectionConfig = provider.getProviderConnectionConfig(null);
+            Assertions.assertArrayEquals("privateKey".getBytes(), providerConnectionConfig.getPrivateKey());
+            Assertions.assertArrayEquals("publicKey".getBytes(), providerConnectionConfig.getKeyCertChain());
+            Assertions.assertArrayEquals("trustCerts".getBytes(), providerConnectionConfig.getTrustCert());
+            Assertions.assertEquals(AuthPolicy.NONE, providerConnectionConfig.getAuthPolicy());
+
+            frameworkModel.destroy();
+        }
+    }
+
+    @Test
+    void getConsumerConnectionConfigTest() {
+        AtomicReference<DubboCertManager> reference = new AtomicReference<>();
+        try (MockedConstruction<DubboCertManager> construction =
+                 Mockito.mockConstruction(DubboCertManager.class, (mock, context) -> {
+                     reference.set(mock);
+                 })) {
+            FrameworkModel frameworkModel = new FrameworkModel();
+            DubboCertProvider provider = new DubboCertProvider(frameworkModel);
+            Assertions.assertNull(provider.getConsumerConnectionConfig(null));
+
+            CertPair certPair = new CertPair("privateKey", "publicKey", "trustCerts", 12345);
+            Mockito.when(reference.get().generateCert()).thenReturn(certPair);
+            Cert connectionConfig = provider.getConsumerConnectionConfig(null);
+            Assertions.assertArrayEquals("privateKey".getBytes(), connectionConfig.getPrivateKey());
+            Assertions.assertArrayEquals("publicKey".getBytes(), connectionConfig.getKeyCertChain());
+            Assertions.assertArrayEquals("trustCerts".getBytes(), connectionConfig.getTrustCert());
+
+            frameworkModel.destroy();
+        }
+    }
+}

dubbo-plugin/dubbo-security/src/test/resources/certs/broken-ca.crt

@@ -0,0 +1,7 @@
+-----BEGIN CERTIFICATE-----
+MIICSjCCAbOgAwIBAgIJAJHGGR4dGioHMA0GCSqGSIb3DQEBCwUAMFYxCzAJBgNV
+HQ8BAf8EBAMCAgQwDQYJKoZIhvcNAQELBQADgYEAHzC7jdYlzAVmddi/gdAeKPau
+sPBG/C2HCWqHzpCUHcKuvMzDVkY/MP2o6JIW2DBbY64bO/FceExhjcykgaYtCH/m
+oIU63+CFOTtR7otyQAWHqXa7q4SbCDlG7DyRFxqG0txPtGvy12lgldA2+RgcigQG
+Dfcog5wrJytaQ6UA0wE=
+-----END CERTIFICATE-----

dubbo-plugin/dubbo-security/src/test/resources/certs/ca.crt

@@ -0,0 +1,15 @@
+-----BEGIN CERTIFICATE-----
+MIICSjCCAbOgAwIBAgIJAJHGGR4dGioHMA0GCSqGSIb3DQEBCwUAMFYxCzAJBgNV
+BAYTAkFVMRMwEQYDVQQIEwpTb21lLVN0YXRlMSEwHwYDVQQKExhJbnRlcm5ldCBX
+aWRnaXRzIFB0eSBMdGQxDzANBgNVBAMTBnRlc3RjYTAeFw0xNDExMTEyMjMxMjla
+Fw0yNDExMDgyMjMxMjlaMFYxCzAJBgNVBAYTAkFVMRMwEQYDVQQIEwpTb21lLVN0
+YXRlMSEwHwYDVQQKExhJbnRlcm5ldCBXaWRnaXRzIFB0eSBMdGQxDzANBgNVBAMT
+BnRlc3RjYTCBnzANBgkqhkiG9w0BAQEFAAOBjQAwgYkCgYEAwEDfBV5MYdlHVHJ7
++L4nxrZy7mBfAVXpOc5vMYztssUI7mL2/iYujiIXM+weZYNTEpLdjyJdu7R5gGUu
+g1jSVK/EPHfc74O7AyZU34PNIP4Sh33N+/A5YexrNgJlPY+E3GdVYi4ldWJjgkAd
+Qah2PH5ACLrIIC6tRka9hcaBlIECAwEAAaMgMB4wDAYDVR0TBAUwAwEB/zAOBgNV
+HQ8BAf8EBAMCAgQwDQYJKoZIhvcNAQELBQADgYEAHzC7jdYlzAVmddi/gdAeKPau
+sPBG/C2HCWqHzpCUHcKuvMzDVkY/MP2o6JIW2DBbY64bO/FceExhjcykgaYtCH/m
+oIU63+CFOTtR7otyQAWHqXa7q4SbCDlG7DyRFxqG0txPtGvy12lgldA2+RgcigQG
+Dfcog5wrJytaQ6UA0wE=
+-----END CERTIFICATE-----

dubbo-plugin/dubbo-security/src/test/resources/certs/token

@@ -0,0 +1 @@
+tokentokentoken

dubbo-plugin/pom.xml

@@ -32,6 +32,7 @@
         <module>dubbo-qos</module>
         <module>dubbo-auth</module>
         <module>dubbo-reactive</module>
+        <module>dubbo-security</module>
         <module>dubbo-spring-security</module>
     </modules>
     <properties>

dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyClient.java

@@ -28,8 +28,9 @@ import org.apache.dubbo.common.utils.StringUtils;
 import org.apache.dubbo.remoting.ChannelHandler;
 import org.apache.dubbo.remoting.Constants;
 import org.apache.dubbo.remoting.RemotingException;
-import org.apache.dubbo.remoting.transport.netty4.ssl.SslClientTlsHandler;
 import org.apache.dubbo.remoting.transport.AbstractClient;
+import org.apache.dubbo.remoting.transport.netty4.ssl.SslClientTlsHandler;
+import org.apache.dubbo.remoting.transport.netty4.ssl.SslContexts;
 import org.apache.dubbo.remoting.utils.UrlUtils;
 
 import io.netty.bootstrap.Bootstrap;
@@ -41,13 +42,13 @@ import io.netty.channel.ChannelOption;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.proxy.Socks5ProxyHandler;
+import io.netty.handler.ssl.SslContext;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.util.concurrent.EventExecutorGroup;
 
 import java.net.InetSocketAddress;
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import static org.apache.dubbo.common.constants.CommonConstants.SSL_ENABLED_KEY;
 import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_CLIENT_CONNECT_TIMEOUT;
 import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_CONNECT_PROVIDER;
 import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_DISCONNECT_PROVIDER;
@@ -119,14 +120,15 @@ public class NettyClient extends AbstractClient {
             .channel(socketChannelClass());
 
         bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Math.max(DEFAULT_CONNECT_TIMEOUT, getConnectTimeout()));
+        SslContext sslContext = SslContexts.buildClientSslContext(getUrl());
         bootstrap.handler(new ChannelInitializer<SocketChannel>() {
 
             @Override
             protected void initChannel(SocketChannel ch) throws Exception {
                 int heartbeatInterval = UrlUtils.getHeartbeat(getUrl());
 
-                if (getUrl().getParameter(SSL_ENABLED_KEY, false)) {
+                if (sslContext != null) {
-                    ch.pipeline().addLast("negotiation", new SslClientTlsHandler(getUrl()));
+                    ch.pipeline().addLast("negotiation", new SslClientTlsHandler(sslContext));
                 }
 
                 NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this);

dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyConnectionClient.java

@@ -28,6 +28,7 @@ import org.apache.dubbo.remoting.RemotingException;
 import org.apache.dubbo.remoting.api.WireProtocol;
 import org.apache.dubbo.remoting.api.connection.AbstractConnectionClient;
 import org.apache.dubbo.remoting.transport.netty4.ssl.SslClientTlsHandler;
+import org.apache.dubbo.remoting.transport.netty4.ssl.SslContexts;
 
 import io.netty.bootstrap.Bootstrap;
 import io.netty.buffer.PooledByteBufAllocator;
@@ -38,6 +39,7 @@ import io.netty.channel.ChannelOption;
 import io.netty.channel.ChannelPipeline;
 import io.netty.channel.EventLoop;
 import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.ssl.SslContext;
 import io.netty.util.AttributeKey;
 import io.netty.util.concurrent.DefaultPromise;
 import io.netty.util.concurrent.GlobalEventExecutor;
@@ -48,7 +50,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_CLIENT_THREADPOOL;
-import static org.apache.dubbo.common.constants.CommonConstants.SSL_ENABLED_KEY;
 import static org.apache.dubbo.common.constants.CommonConstants.THREADPOOL_KEY;
 import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_CLIENT_CONNECT_TIMEOUT;
 import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_CONNECT_PROVIDER;
@@ -107,6 +108,7 @@ public class NettyConnectionClient extends AbstractConnectionClient {
 
         final NettyConnectionHandler connectionHandler = new NettyConnectionHandler(this);
         nettyBootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getConnectTimeout());
+        SslContext sslContext = SslContexts.buildClientSslContext(getUrl());
         nettyBootstrap.handler(new ChannelInitializer<SocketChannel>() {
             @Override
             protected void initChannel(SocketChannel ch) {
@@ -114,8 +116,8 @@ public class NettyConnectionClient extends AbstractConnectionClient {
                 final ChannelPipeline pipeline = ch.pipeline();
                 NettySslContextOperator nettySslContextOperator = new NettySslContextOperator();
 
-                if (getUrl().getParameter(SSL_ENABLED_KEY, false)) {
+                if (sslContext != null) {
-                    pipeline.addLast("negotiation", new SslClientTlsHandler(getUrl()));
+                    pipeline.addLast("negotiation", new SslClientTlsHandler(sslContext));
                 }
 
 //                pipeline.addLast("logging", new LoggingHandler(LogLevel.INFO)); //for debug

dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServer.java

@@ -26,7 +26,6 @@ import org.apache.dubbo.remoting.Channel;
 import org.apache.dubbo.remoting.ChannelHandler;
 import org.apache.dubbo.remoting.Constants;
 import org.apache.dubbo.remoting.RemotingException;
-import org.apache.dubbo.remoting.transport.netty4.ssl.SslContexts;
 import org.apache.dubbo.remoting.api.WireProtocol;
 import org.apache.dubbo.remoting.api.pu.AbstractPortUnificationServer;
 import org.apache.dubbo.remoting.transport.dispatcher.ChannelHandlers;
@@ -39,7 +38,6 @@ import io.netty.channel.ChannelOption;
 import io.netty.channel.ChannelPipeline;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
-import io.netty.handler.ssl.SslContext;
 import io.netty.util.concurrent.Future;
 
 import java.net.InetSocketAddress;
@@ -52,7 +50,6 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static org.apache.dubbo.common.constants.CommonConstants.ANYHOST_KEY;
 import static org.apache.dubbo.common.constants.CommonConstants.ANYHOST_VALUE;
 import static org.apache.dubbo.common.constants.CommonConstants.IO_THREADS_KEY;
-import static org.apache.dubbo.common.constants.CommonConstants.SSL_ENABLED_KEY;
 import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_CLOSE;
 import static org.apache.dubbo.remoting.Constants.EVENT_LOOP_BOSS_POOL_NAME;
 import static org.apache.dubbo.remoting.Constants.EVENT_LOOP_WORKER_POOL_NAME;
@@ -115,13 +112,6 @@ public class NettyPortUnificationServer extends AbstractPortUnificationServer {
             getUrl().getPositiveParameter(IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS),
             EVENT_LOOP_WORKER_POOL_NAME);
 
-        final boolean enableSsl = getUrl().getParameter(SSL_ENABLED_KEY, false);
-        final SslContext sslContext;
-        if (enableSsl) {
-            sslContext = SslContexts.buildServerSslContext(getUrl());
-        } else {
-            sslContext = null;
-        }
         bootstrap.group(bossGroup, workerGroup)
             .channel(NettyEventLoopFactory.serverSocketChannelClass())
             .option(ChannelOption.SO_REUSEADDR, Boolean.TRUE)
@@ -133,7 +123,7 @@ public class NettyPortUnificationServer extends AbstractPortUnificationServer {
                     // Do not add idle state handler here, because it should be added in the protocol handler.
                     final ChannelPipeline p = ch.pipeline();
                     final NettyPortUnificationServerHandler puHandler;
-                    puHandler = new NettyPortUnificationServerHandler(getUrl(), sslContext, true, getProtocols(),
+                    puHandler = new NettyPortUnificationServerHandler(getUrl(), true, getProtocols(),
                         NettyPortUnificationServer.this, NettyPortUnificationServer.this.dubboChannels,
                         getSupportedUrls(), getSupportedHandlers());
                     p.addLast("negotiation-protocol", puHandler);

dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServerHandler.java

@@ -20,12 +20,15 @@ import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.io.Bytes;
 import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
 import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.ssl.CertManager;
+import org.apache.dubbo.common.ssl.ProviderCert;
 import org.apache.dubbo.common.utils.NetUtils;
 import org.apache.dubbo.remoting.Channel;
 import org.apache.dubbo.remoting.ChannelHandler;
 import org.apache.dubbo.remoting.api.ProtocolDetector;
 import org.apache.dubbo.remoting.api.WireProtocol;
 import org.apache.dubbo.remoting.buffer.ChannelBuffer;
+import org.apache.dubbo.remoting.transport.netty4.ssl.SslContexts;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
@@ -33,7 +36,9 @@ import io.netty.channel.ChannelPipeline;
 import io.netty.handler.codec.ByteToMessageDecoder;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslHandler;
+import io.netty.handler.ssl.SslHandshakeCompletionEvent;
 
+import javax.net.ssl.SSLSession;
 import java.net.InetSocketAddress;
 import java.util.List;
 import java.util.Map;
@@ -45,8 +50,6 @@ public class NettyPortUnificationServerHandler extends ByteToMessageDecoder {
 
     private static final ErrorTypeAwareLogger LOGGER = LoggerFactory.getErrorTypeAwareLogger(
         NettyPortUnificationServerHandler.class);
-
-    private final SslContext sslCtx;
     private final URL url;
     private final ChannelHandler handler;
     private final boolean detectSsl;
@@ -56,11 +59,10 @@ public class NettyPortUnificationServerHandler extends ByteToMessageDecoder {
     private final Map<String, ChannelHandler> handlerMapper;
 
 
-    public NettyPortUnificationServerHandler(URL url, SslContext sslCtx, boolean detectSsl,
+    public NettyPortUnificationServerHandler(URL url, boolean detectSsl,
                                              List<WireProtocol> protocols, ChannelHandler handler,
                                              Map<String, Channel> dubboChannels, Map<String, URL> urlMapper, Map<String, ChannelHandler> handlerMapper) {
         this.url = url;
-        this.sslCtx = sslCtx;
         this.protocols = protocols;
         this.detectSsl = detectSsl;
         this.handler = handler;
@@ -84,6 +86,21 @@ public class NettyPortUnificationServerHandler extends ByteToMessageDecoder {
         }
     }
 
+    @Override
+    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+        if (evt instanceof SslHandshakeCompletionEvent) {
+            SslHandshakeCompletionEvent handshakeEvent = (SslHandshakeCompletionEvent) evt;
+            if (handshakeEvent.isSuccess()) {
+                SSLSession session = ctx.pipeline().get(SslHandler.class).engine().getSession();
+                LOGGER.info("TLS negotiation succeed with session: " + session);
+            } else {
+                LOGGER.error(INTERNAL_ERROR, "", "", "TLS negotiation failed when trying to accept new connection.", handshakeEvent.cause());
+                ctx.close();
+            }
+        }
+        super.userEventTriggered(ctx, evt);
+    }
+
     @Override
     protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
         throws Exception {
@@ -94,8 +111,11 @@ public class NettyPortUnificationServerHandler extends ByteToMessageDecoder {
             return;
         }
 
-        if (isSsl(in)) {
+        CertManager certManager = url.getOrDefaultFrameworkModel().getBeanFactory().getBean(CertManager.class);
-            enableSsl(ctx);
+        ProviderCert providerConnectionConfig = certManager.getProviderConnectionConfig(url, ctx.channel().remoteAddress());
+
+        if (providerConnectionConfig != null && isSsl(in)) {
+            enableSsl(ctx, providerConnectionConfig);
         } else {
             for (final WireProtocol protocol : protocols) {
                 in.markReaderIndex();
@@ -136,13 +156,12 @@ public class NettyPortUnificationServerHandler extends ByteToMessageDecoder {
         }
     }
 
-    private void enableSsl(ChannelHandlerContext ctx) {
+    private void enableSsl(ChannelHandlerContext ctx, ProviderCert providerConnectionConfig) {
         ChannelPipeline p = ctx.pipeline();
-        if (sslCtx != null) {
+        SslContext sslContext = SslContexts.buildServerSslContext(providerConnectionConfig);
-            p.addLast("ssl", sslCtx.newHandler(ctx.alloc()));
+        p.addLast("ssl", sslContext.newHandler(ctx.alloc()));
-        }
         p.addLast("unificationA",
-            new NettyPortUnificationServerHandler(url, sslCtx, false, protocols,
+            new NettyPortUnificationServerHandler(url, false, protocols,
                 handler, dubboChannels, urlMapper, handlerMapper));
         p.remove(this);
     }

dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyServer.java

@@ -49,7 +49,6 @@ import java.util.Map;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static org.apache.dubbo.common.constants.CommonConstants.IO_THREADS_KEY;
 import static org.apache.dubbo.common.constants.CommonConstants.KEEP_ALIVE_KEY;
-import static org.apache.dubbo.common.constants.CommonConstants.SSL_ENABLED_KEY;
 import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_CLOSE;
 import static org.apache.dubbo.remoting.Constants.EVENT_LOOP_BOSS_POOL_NAME;
 import static org.apache.dubbo.remoting.Constants.EVENT_LOOP_WORKER_POOL_NAME;
@@ -128,7 +127,6 @@ public class NettyServer extends AbstractServer {
 
     protected void initServerBootstrap(NettyServerHandler nettyServerHandler) {
         boolean keepalive = getUrl().getParameter(KEEP_ALIVE_KEY, Boolean.FALSE);
-
         bootstrap.group(bossGroup, workerGroup)
             .channel(NettyEventLoopFactory.serverSocketChannelClass())
             .option(ChannelOption.SO_REUSEADDR, Boolean.TRUE)
@@ -141,9 +139,7 @@ public class NettyServer extends AbstractServer {
                     // FIXME: should we use getTimeout()?
                     int idleTimeout = UrlUtils.getIdleTimeout(getUrl());
                     NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
-                    if (getUrl().getParameter(SSL_ENABLED_KEY, false)) {
                     ch.pipeline().addLast("negotiation", new SslServerTlsHandler(getUrl()));
-                    }
                     ch.pipeline()
                         .addLast("decoder", adapter.getDecoder())
                         .addLast("encoder", adapter.getEncoder())

dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/ssl/SslClientTlsHandler.java

@@ -58,7 +58,7 @@ public class SslClientTlsHandler extends ChannelInboundHandlerAdapter {
             SslHandshakeCompletionEvent handshakeEvent = (SslHandshakeCompletionEvent) evt;
             if (handshakeEvent.isSuccess()) {
                 SSLSession session = ctx.pipeline().get(SslHandler.class).engine().getSession();
-                logger.info("TLS negotiation succeed with session: " + session);
+                logger.info("TLS negotiation succeed with: " + session.getPeerHost());
                 ctx.pipeline().remove(this);
             } else {
                 logger.error(INTERNAL_ERROR, "unknown error in remoting module", "", "TLS negotiation failed when trying to accept new connection.", handshakeEvent.cause());

dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/ssl/SslContexts.java

@@ -19,8 +19,10 @@ package org.apache.dubbo.remoting.transport.netty4.ssl;
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
 import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.config.SslConfig;
+import org.apache.dubbo.common.ssl.AuthPolicy;
-import org.apache.dubbo.config.context.ConfigManager;
+import org.apache.dubbo.common.ssl.Cert;
+import org.apache.dubbo.common.ssl.CertManager;
+import org.apache.dubbo.common.ssl.ProviderCert;
 
 import io.netty.handler.ssl.ClientAuth;
 import io.netty.handler.ssl.OpenSsl;
@@ -40,19 +42,16 @@ public class SslContexts {
 
     private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(SslContexts.class);
 
-    public static SslContext buildServerSslContext(URL url) {
+    public static SslContext buildServerSslContext(ProviderCert providerConnectionConfig) {
-        ConfigManager globalConfigManager = url.getOrDefaultApplicationModel().getApplicationConfigManager();
-        SslConfig sslConfig = globalConfigManager.getSsl().orElseThrow(() -> new IllegalStateException("Ssl enabled, but no ssl cert information provided!"));
-
         SslContextBuilder sslClientContextBuilder;
         InputStream serverKeyCertChainPathStream = null;
         InputStream serverPrivateKeyPathStream = null;
         InputStream serverTrustCertStream = null;
         try {
-            serverKeyCertChainPathStream = sslConfig.getServerKeyCertChainPathStream();
+            serverKeyCertChainPathStream = providerConnectionConfig.getKeyCertChainInputStream();
-            serverPrivateKeyPathStream = sslConfig.getServerPrivateKeyPathStream();
+            serverPrivateKeyPathStream = providerConnectionConfig.getPrivateKeyInputStream();
-            serverTrustCertStream = sslConfig.getServerTrustCertCollectionPathStream();
+            serverTrustCertStream = providerConnectionConfig.getTrustCertInputStream();
-            String password = sslConfig.getServerKeyPassword();
+            String password = providerConnectionConfig.getPassword();
             if (password != null) {
                 sslClientContextBuilder = SslContextBuilder.forServer(serverKeyCertChainPathStream,
                     serverPrivateKeyPathStream, password);
@@ -63,7 +62,11 @@ public class SslContexts {
 
             if (serverTrustCertStream != null) {
                 sslClientContextBuilder.trustManager(serverTrustCertStream);
+                if (providerConnectionConfig.getAuthPolicy() == AuthPolicy.CLIENT_AUTH) {
                     sslClientContextBuilder.clientAuth(ClientAuth.REQUIRE);
+                } else {
+                    sslClientContextBuilder.clientAuth(ClientAuth.OPTIONAL);
+                }
             }
         } catch (Exception e) {
             throw new IllegalArgumentException("Could not find certificate file or the certificate is invalid.", e);
@@ -80,23 +83,26 @@ public class SslContexts {
     }
 
     public static SslContext buildClientSslContext(URL url) {
-        ConfigManager globalConfigManager = url.getOrDefaultApplicationModel().getApplicationConfigManager();
+        CertManager certManager = url.getOrDefaultFrameworkModel().getBeanFactory().getBean(CertManager.class);
-        SslConfig sslConfig = globalConfigManager.getSsl().orElseThrow(() -> new IllegalStateException("Ssl enabled, but no ssl cert information provided!"));
+        Cert consumerConnectionConfig = certManager.getConsumerConnectionConfig(url);
+        if (consumerConnectionConfig == null) {
+            return null;
+        }
 
         SslContextBuilder builder = SslContextBuilder.forClient();
         InputStream clientTrustCertCollectionPath = null;
         InputStream clientCertChainFilePath = null;
         InputStream clientPrivateKeyFilePath = null;
         try {
-            clientTrustCertCollectionPath = sslConfig.getClientTrustCertCollectionPathStream();
+            clientTrustCertCollectionPath = consumerConnectionConfig.getTrustCertInputStream();
             if (clientTrustCertCollectionPath != null) {
                 builder.trustManager(clientTrustCertCollectionPath);
             }
 
-            clientCertChainFilePath = sslConfig.getClientKeyCertChainPathStream();
+            clientCertChainFilePath = consumerConnectionConfig.getKeyCertChainInputStream();
-            clientPrivateKeyFilePath = sslConfig.getClientPrivateKeyPathStream();
+            clientPrivateKeyFilePath = consumerConnectionConfig.getPrivateKeyInputStream();
             if (clientCertChainFilePath != null && clientPrivateKeyFilePath != null) {
-                String password = sslConfig.getClientKeyPassword();
+                String password = consumerConnectionConfig.getPassword();
                 if (password != null) {
                     builder.keyManager(clientCertChainFilePath, clientPrivateKeyFilePath, password);
                 } else {

dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/ssl/SslServerTlsHandler.java

@@ -19,6 +19,9 @@ package org.apache.dubbo.remoting.transport.netty4.ssl;
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
 import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.ssl.AuthPolicy;
+import org.apache.dubbo.common.ssl.CertManager;
+import org.apache.dubbo.common.ssl.ProviderCert;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
@@ -36,25 +39,18 @@ import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_ERR
 public class SslServerTlsHandler extends ByteToMessageDecoder {
     private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(SslServerTlsHandler.class);
 
-    private final SslContext sslContext;
+    private final URL url;
-    private final boolean detectSsl;
 
-
+    private final boolean sslDetected;
-    public SslServerTlsHandler() {
-        this(null, false);
-    }
 
     public SslServerTlsHandler(URL url) {
-        this(SslContexts.buildServerSslContext(url));
+        this.url = url;
+        this.sslDetected = false;
     }
 
-    public SslServerTlsHandler(SslContext sslContext) {
+    public SslServerTlsHandler(URL url, boolean sslDetected) {
-        this(sslContext, true);
+        this.url = url;
-    }
+        this.sslDetected = sslDetected;
-
-    public SslServerTlsHandler(SslContext sslContext, boolean detectSsl) {
-        this.sslContext = sslContext;
-        this.detectSsl = detectSsl;
     }
 
     @Override
@@ -68,7 +64,7 @@ public class SslServerTlsHandler extends ByteToMessageDecoder {
             SslHandshakeCompletionEvent handshakeEvent = (SslHandshakeCompletionEvent) evt;
             if (handshakeEvent.isSuccess()) {
                 SSLSession session = ctx.pipeline().get(SslHandler.class).engine().getSession();
-                logger.info("TLS negotiation succeed with session: " + session);
+                logger.info("TLS negotiation succeed with: " + session.getPeerHost());
                 // Remove after handshake success.
                 ctx.pipeline().remove(this);
             } else {
@@ -86,23 +82,42 @@ public class SslServerTlsHandler extends ByteToMessageDecoder {
             return;
         }
 
-        if (isSsl(byteBuf)) {
+        if (sslDetected) {
-            enableSsl(channelHandlerContext);
+            return;
         }
 
+        CertManager certManager = url.getOrDefaultFrameworkModel().getBeanFactory().getBean(CertManager.class);
+        ProviderCert providerConnectionConfig = certManager.getProviderConnectionConfig(url, channelHandlerContext.channel().remoteAddress());
+
+        if (providerConnectionConfig == null) {
+            ChannelPipeline p = channelHandlerContext.pipeline();
+            p.remove(this);
+            return;
+        }
+
+        if (isSsl(byteBuf)) {
+            SslContext sslContext = SslContexts.buildServerSslContext(providerConnectionConfig);
+            enableSsl(channelHandlerContext, sslContext);
+            return;
+        }
+
+        if (providerConnectionConfig.getAuthPolicy() == AuthPolicy.NONE) {
+            ChannelPipeline p = channelHandlerContext.pipeline();
+            p.remove(this);
+        }
+
+        logger.error(INTERNAL_ERROR, "", "", "TLS negotiation failed when trying to accept new connection.");
+        channelHandlerContext.close();
     }
 
     private boolean isSsl(ByteBuf buf) {
-        if (detectSsl) {
         return SslHandler.isEncrypted(buf);
     }
-        return false;
-    }
 
-    private void enableSsl(ChannelHandlerContext ctx) {
+    private void enableSsl(ChannelHandlerContext ctx, SslContext sslContext) {
         ChannelPipeline p = ctx.pipeline();
         ctx.pipeline().addAfter(ctx.name(), null, sslContext.newHandler(ctx.alloc()));
-        p.addLast("unificationA", new SslServerTlsHandler(sslContext, false));
+        p.addLast("unificationA", new SslServerTlsHandler(url, true));
         p.remove(this);
     }
 

dubbo-test/dubbo-dependencies-all/pom.xml

@@ -194,6 +194,10 @@
             <groupId>org.apache.dubbo</groupId>
             <artifactId>dubbo-auth</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-security</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.apache.dubbo</groupId>
             <artifactId>dubbo-qos</artifactId>