This closes #4088

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/ActiveMQClientMessageBundle.java

@@ -241,4 +241,7 @@ public interface ActiveMQClientMessageBundle {
 
    @Message(id = 219066, value = "The connection was redirected")
    ActiveMQRoutingException redirected();
+
+   @Message(id = 219067, value = "Keystore alias {0} not found in {1}", format = Message.Format.MESSAGE_FORMAT)
+   IllegalArgumentException keystoreAliasNotFound(String keystoreAlias, String keystorePath);
 }

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java

@@ -236,6 +236,8 @@ public class NettyConnector extends AbstractConnector {
 
    private String keyStorePassword;
 
+   private String keyStoreAlias;
+
    private String trustStoreProvider;
 
    private String trustStoreType;
@@ -399,6 +401,8 @@ public class NettyConnector extends AbstractConnector {
 
          keyStorePassword = ConfigurationHelper.getPasswordProperty(TransportConstants.KEYSTORE_PASSWORD_PROP_NAME, TransportConstants.DEFAULT_KEYSTORE_PASSWORD, configuration, ActiveMQDefaultConfiguration.getPropMaskPassword(), ActiveMQDefaultConfiguration.getPropPasswordCodec());
 
+         keyStoreAlias = ConfigurationHelper.getStringProperty(TransportConstants.KEYSTORE_ALIAS_PROP_NAME, TransportConstants.DEFAULT_KEYSTORE_ALIAS, configuration);
+
          trustStoreProvider = ConfigurationHelper.getStringProperty(TransportConstants.TRUSTSTORE_PROVIDER_PROP_NAME, TransportConstants.DEFAULT_TRUSTSTORE_PROVIDER, configuration);
 
          trustStoreType = ConfigurationHelper.getStringProperty(TransportConstants.TRUSTSTORE_TYPE_PROP_NAME, TransportConstants.DEFAULT_TRUSTSTORE_TYPE, configuration);
@@ -431,6 +435,7 @@ public class NettyConnector extends AbstractConnector {
          keyStoreType = TransportConstants.DEFAULT_KEYSTORE_TYPE;
          keyStorePath = TransportConstants.DEFAULT_KEYSTORE_PATH;
          keyStorePassword = TransportConstants.DEFAULT_KEYSTORE_PASSWORD;
+         keyStoreAlias = TransportConstants.DEFAULT_KEYSTORE_ALIAS;
          trustStoreProvider = TransportConstants.DEFAULT_TRUSTSTORE_PROVIDER;
          trustStoreType = TransportConstants.DEFAULT_TRUSTSTORE_TYPE;
          trustStorePath = TransportConstants.DEFAULT_TRUSTSTORE_PATH;
@@ -567,6 +572,7 @@ public class NettyConnector extends AbstractConnector {
       final String realKeyStoreProvider;
       final String realKeyStoreType;
       final String realKeyStorePassword;
+      final String realKeyStoreAlias;
       final String realTrustStorePath;
       final String realTrustStoreProvider;
       final String realTrustStoreType;
@@ -578,6 +584,7 @@ public class NettyConnector extends AbstractConnector {
             realKeyStoreProvider = keyStoreProvider;
             realKeyStoreType = keyStoreType;
             realKeyStorePassword = keyStorePassword;
+            realKeyStoreAlias = keyStoreAlias;
             realTrustStorePath = trustStorePath;
             realTrustStoreProvider = trustStoreProvider;
             realTrustStoreType = trustStoreType;
@@ -585,6 +592,7 @@ public class NettyConnector extends AbstractConnector {
          } else {
             realKeyStorePath = Stream.of(System.getProperty(ACTIVEMQ_KEYSTORE_PATH_PROP_NAME), System.getProperty(JAVAX_KEYSTORE_PATH_PROP_NAME), keyStorePath).map(v -> useDefaultSslContext ? keyStorePath : v).filter(Objects::nonNull).findFirst().orElse(null);
             realKeyStorePassword = Stream.of(System.getProperty(ACTIVEMQ_KEYSTORE_PASSWORD_PROP_NAME), System.getProperty(JAVAX_KEYSTORE_PASSWORD_PROP_NAME), keyStorePassword).map(v -> useDefaultSslContext ? keyStorePassword : v).filter(Objects::nonNull).findFirst().orElse(null);
+            realKeyStoreAlias = keyStoreAlias;
 
             Pair<String, String> keyStoreCompat = SSLSupport.getValidProviderAndType(Stream.of(System.getProperty(ACTIVEMQ_KEYSTORE_PROVIDER_PROP_NAME), System.getProperty(JAVAX_KEYSTORE_PROVIDER_PROP_NAME), keyStoreProvider).map(v -> useDefaultSslContext ? keyStoreProvider : v).filter(Objects::nonNull).findFirst().orElse(null),
                                                                                      Stream.of(System.getProperty(ACTIVEMQ_KEYSTORE_TYPE_PROP_NAME), System.getProperty(JAVAX_KEYSTORE_TYPE_PROP_NAME), keyStoreType).map(v -> useDefaultSslContext ? keyStoreType : v).filter(Objects::nonNull).findFirst().orElse(null));
@@ -604,6 +612,7 @@ public class NettyConnector extends AbstractConnector {
          realKeyStoreProvider = null;
          realKeyStoreType = null;
          realKeyStorePassword = null;
+         realKeyStoreAlias = null;
          realTrustStorePath = null;
          realTrustStoreProvider = null;
          realTrustStoreType = null;
@@ -645,6 +654,7 @@ public class NettyConnector extends AbstractConnector {
                   .keystorePath(realKeyStorePath)
                   .keystoreType(realKeyStoreType)
                   .keystorePassword(realKeyStorePassword)
+                  .keystoreAlias(realKeyStoreAlias)
                   .truststoreProvider(realTrustStoreProvider)
                   .truststorePath(realTrustStorePath)
                   .truststoreType(realTrustStoreType)

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java

@@ -97,6 +97,8 @@ public class TransportConstants {
 
    public static final String KEYSTORE_PASSWORD_PROP_NAME = "keyStorePassword";
 
+   public static final String KEYSTORE_ALIAS_PROP_NAME = "keyStoreAlias";
+
    public static final String TRUSTSTORE_PROVIDER_PROP_NAME = "trustStoreProvider";
 
    public static final String TRUSTSTORE_TYPE_PROP_NAME = "trustStoreType";
@@ -258,6 +260,8 @@ public class TransportConstants {
 
    public static final boolean DEFAULT_USE_DEFAULT_SSL_CONTEXT = false;
 
+   public static final String DEFAULT_KEYSTORE_ALIAS = null;
+
    public static final boolean DEFAULT_TCP_NODELAY = true;
 
    public static final int DEFAULT_TCP_SENDBUFFER_SIZE = 1024 * 1024;
@@ -400,6 +404,7 @@ public class TransportConstants {
       allowableAcceptorKeys.add(TransportConstants.KEYSTORE_TYPE_PROP_NAME);
       allowableAcceptorKeys.add(TransportConstants.KEYSTORE_PATH_PROP_NAME);
       allowableAcceptorKeys.add(TransportConstants.KEYSTORE_PASSWORD_PROP_NAME);
+      allowableAcceptorKeys.add(TransportConstants.KEYSTORE_ALIAS_PROP_NAME);
       allowableAcceptorKeys.add(TransportConstants.TRUSTSTORE_PROVIDER_PROP_NAME);
       allowableAcceptorKeys.add(TransportConstants.TRUSTSTORE_TYPE_PROP_NAME);
       allowableAcceptorKeys.add(TransportConstants.TRUSTSTORE_PATH_PROP_NAME);
@@ -472,6 +477,7 @@ public class TransportConstants {
       allowableConnectorKeys.add(TransportConstants.KEYSTORE_TYPE_PROP_NAME);
       allowableConnectorKeys.add(TransportConstants.KEYSTORE_PATH_PROP_NAME);
       allowableConnectorKeys.add(TransportConstants.KEYSTORE_PASSWORD_PROP_NAME);
+      allowableConnectorKeys.add(TransportConstants.KEYSTORE_ALIAS_PROP_NAME);
       allowableConnectorKeys.add(TransportConstants.TRUSTSTORE_PROVIDER_PROP_NAME);
       allowableConnectorKeys.add(TransportConstants.TRUSTSTORE_TYPE_PROP_NAME);
       allowableConnectorKeys.add(TransportConstants.TRUSTSTORE_PATH_PROP_NAME);

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/ssl/AliasedKeyManager.java

@@ -0,0 +1,89 @@
+/*
+ *  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.activemq.artemis.core.remoting.impl.ssl;
+
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.X509ExtendedKeyManager;
+import javax.net.ssl.X509KeyManager;
+import java.net.Socket;
+import java.security.Principal;
+import java.security.PrivateKey;
+import java.security.cert.X509Certificate;
+
+public final class AliasedKeyManager extends X509ExtendedKeyManager {
+
+   private X509KeyManager wrapped;
+   private String keystoreAlias;
+
+   public AliasedKeyManager(X509KeyManager wrapped, String keystoreAlias) {
+      super();
+      this.wrapped = wrapped;
+      this.keystoreAlias = keystoreAlias;
+   }
+
+   @Override
+   public String chooseServerAlias(String keyType, Principal[] issuers, Socket socket) {
+      if (keystoreAlias != null) {
+         return keystoreAlias;
+      }
+
+      return wrapped.chooseServerAlias(keyType, issuers, socket);
+   }
+
+   @Override
+   public String chooseEngineServerAlias(String keyType, Principal[] issuers, SSLEngine engine) {
+      if (keystoreAlias != null) {
+         return keystoreAlias;
+      }
+
+      return super.chooseEngineServerAlias(keyType, issuers, engine);
+   }
+
+   @Override
+   public String chooseClientAlias(String[] keyType, Principal[] issuers, Socket socket) {
+      if (keystoreAlias != null) {
+         return keystoreAlias;
+      }
+
+      return wrapped.chooseClientAlias(keyType, issuers, socket);
+   }
+
+   @Override
+   public X509Certificate[] getCertificateChain(String alias) {
+      return wrapped.getCertificateChain(alias);
+   }
+
+   @Override
+   public String[] getClientAliases(String keyType, Principal[] issuers) {
+      return wrapped.getClientAliases(keyType, issuers);
+   }
+
+   @Override
+   public String[] getServerAliases(String keyType, Principal[] issuers) {
+      return wrapped.getServerAliases(keyType, issuers);
+   }
+
+   @Override
+   public PrivateKey getPrivateKey(String alias) {
+      return wrapped.getPrivateKey(alias);
+   }
+
+   @Override
+   public String chooseEngineClientAlias(String[] keyType, Principal[] issuers, SSLEngine engine) {
+      return chooseClientAlias(keyType, issuers, null);
+   }
+}

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/ssl/SSLSupport.java

@@ -22,6 +22,7 @@ import javax.net.ssl.KeyManagerFactory;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.TrustManager;
 import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509KeyManager;
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
@@ -29,15 +30,21 @@ import java.net.MalformedURLException;
 import java.net.URL;
 import java.security.AccessController;
 import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.PrivateKey;
 import java.security.PrivilegedAction;
 import java.security.SecureRandom;
 import java.security.Security;
+import java.security.UnrecoverableKeyException;
 import java.security.cert.CRL;
 import java.security.cert.CertStore;
+import java.security.cert.Certificate;
 import java.security.cert.CertificateFactory;
 import java.security.cert.CollectionCertStoreParameters;
 import java.security.cert.PKIXBuilderParameters;
 import java.security.cert.X509CertSelector;
+import java.security.cert.X509Certificate;
 import java.util.Collection;
 
 import io.netty.handler.ssl.SslContext;
@@ -47,6 +54,7 @@ import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
 import org.apache.activemq.artemis.api.core.Pair;
 import org.apache.activemq.artemis.api.core.TrustManagerFactoryPlugin;
 import org.apache.activemq.artemis.core.client.ActiveMQClientLogger;
+import org.apache.activemq.artemis.core.client.ActiveMQClientMessageBundle;
 import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants;
 import org.apache.activemq.artemis.spi.core.remoting.ssl.SSLContextConfig;
 import org.apache.activemq.artemis.utils.ClassloadingUtil;
@@ -72,6 +80,7 @@ public class SSLSupport {
    private String sslProvider = TransportConstants.DEFAULT_SSL_PROVIDER;
    private boolean trustAll = TransportConstants.DEFAULT_TRUST_ALL;
    private String trustManagerFactoryPlugin = TransportConstants.DEFAULT_TRUST_MANAGER_FACTORY_PLUGIN;
+   private String keystoreAlias = TransportConstants.DEFAULT_KEYSTORE_ALIAS;
 
    public SSLSupport() {
    }
@@ -88,6 +97,7 @@ public class SSLSupport {
       crlPath = config.getCrlPath();
       trustAll = config.isTrustAll();
       trustManagerFactoryPlugin = config.getTrustManagerFactoryPlugin();
+      keystoreAlias = config.getKeystoreAlias();
    }
 
    public String getKeystoreProvider() {
@@ -126,6 +136,15 @@ public class SSLSupport {
       return this;
    }
 
+   public String getKeystoreAlias() {
+      return keystoreAlias;
+   }
+
+   public SSLSupport setKeystoreAlias(String keystoreAlias) {
+      this.keystoreAlias = keystoreAlias;
+      return this;
+   }
+
    public String getTruststoreProvider() {
       return truststoreProvider;
    }
@@ -208,18 +227,34 @@ public class SSLSupport {
 
    public SslContext createNettyContext() throws Exception {
       KeyStore keyStore = SSLSupport.loadKeystore(keystoreProvider, keystoreType, keystorePath, keystorePassword);
-      KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+      SslContextBuilder sslContextBuilder;
-      keyManagerFactory.init(keyStore, keystorePassword.toCharArray());
+      if (keystoreAlias != null) {
-      return SslContextBuilder.forServer(keyManagerFactory).sslProvider(SslProvider.valueOf(sslProvider)).trustManager(loadTrustManagerFactory()).build();
+         Pair<PrivateKey, X509Certificate[]> privateKeyAndCertChain = getPrivateKeyAndCertChain(keyStore);
+         sslContextBuilder = SslContextBuilder.forServer(privateKeyAndCertChain.getA(), privateKeyAndCertChain.getB());
+      } else {
+         sslContextBuilder = SslContextBuilder.forServer(getKeyManagerFactory(keyStore, keystorePassword == null ? null : keystorePassword.toCharArray()));
+      }
+      return sslContextBuilder
+         .sslProvider(SslProvider.valueOf(sslProvider))
+         .trustManager(loadTrustManagerFactory())
+         .build();
    }
 
    public SslContext createNettyClientContext() throws Exception {
       KeyStore keyStore = SSLSupport.loadKeystore(keystoreProvider, keystoreType, keystorePath, keystorePassword);
-      KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+      SslContextBuilder sslContextBuilder = SslContextBuilder
-      keyManagerFactory.init(keyStore, keystorePassword == null ? null : keystorePassword.toCharArray());
+         .forClient()
-      return SslContextBuilder.forClient().sslProvider(SslProvider.valueOf(sslProvider)).keyManager(keyManagerFactory).trustManager(loadTrustManagerFactory()).build();
+         .sslProvider(SslProvider.valueOf(sslProvider))
-   }
+         .trustManager(loadTrustManagerFactory());
+      if (keystoreAlias != null) {
+         Pair<PrivateKey, X509Certificate[]> privateKeyAndCertChain = getPrivateKeyAndCertChain(keyStore);
+         sslContextBuilder.keyManager(privateKeyAndCertChain.getA(), privateKeyAndCertChain.getB());
+      } else {
+         sslContextBuilder.keyManager(getKeyManagerFactory(keyStore, keystorePassword == null ? null : keystorePassword.toCharArray()));
+      }
 
+      return sslContextBuilder.build();
+   }
 
    public static String[] parseCommaSeparatedListIntoArray(String suites) {
       String[] cipherSuites = suites.split(",");
@@ -321,7 +356,15 @@ public class SSLSupport {
       if (factory == null) {
          return null;
       }
-      return factory.getKeyManagers();
+      KeyManager[] keyManagers = factory.getKeyManagers();
+      if (keystoreAlias != null) {
+         for (int i = 0; i < keyManagers.length; i++) {
+            if (keyManagers[i] instanceof X509KeyManager) {
+               keyManagers[i] = new AliasedKeyManager((X509KeyManager) keyManagers[i], keystoreAlias);
+            }
+         }
+      }
+      return keyManagers;
    }
 
    private KeyManagerFactory loadKeyManagerFactory() throws Exception {
@@ -370,6 +413,24 @@ public class SSLSupport {
       });
    }
 
+   private Pair<PrivateKey, X509Certificate[]> getPrivateKeyAndCertChain(KeyStore keyStore) throws KeyStoreException, UnrecoverableKeyException, NoSuchAlgorithmException {
+      PrivateKey key = (PrivateKey) keyStore.getKey(keystoreAlias, keystorePassword.toCharArray());
+      if (key == null) {
+         throw ActiveMQClientMessageBundle.BUNDLE.keystoreAliasNotFound(keystoreAlias, keystorePath);
+      }
+
+      Certificate[] chain = keyStore.getCertificateChain(keystoreAlias);
+      X509Certificate[] certChain = new X509Certificate[chain.length];
+      System.arraycopy(chain, 0, certChain, 0, chain.length);
+      return new Pair(key, certChain);
+   }
+
+   private KeyManagerFactory getKeyManagerFactory(KeyStore keyStore, char[] keystorePassword) throws NoSuchAlgorithmException, KeyStoreException, UnrecoverableKeyException {
+      KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+      keyManagerFactory.init(keyStore, keystorePassword);
+      return keyManagerFactory;
+   }
+
    /**
     * The changes ARTEMIS-3155 introduced an incompatibility with old clients using the keyStoreProvider and
     * trustStoreProvider URL properties. These old clients use these properties to set the *type* of store

artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/remoting/ssl/SSLContextConfig.java

@@ -40,6 +40,7 @@ public final class SSLContextConfig {
       private String crlPath = TransportConstants.DEFAULT_CRL_PATH;
       private String trustManagerFactoryPlugin = TransportConstants.DEFAULT_TRUST_MANAGER_FACTORY_PLUGIN;
       private boolean trustAll = TransportConstants.DEFAULT_TRUST_ALL;
+      private String keystoreAlias = TransportConstants.DEFAULT_KEYSTORE_ALIAS;
 
       private Builder() {
       }
@@ -58,6 +59,7 @@ public final class SSLContextConfig {
          crlPath = config.getCrlPath();
          truststoreProvider = config.getTruststoreProvider();
          trustAll = config.trustAll;
+         keystoreAlias = config.keystoreAlias;
          return this;
       }
 
@@ -65,7 +67,7 @@ public final class SSLContextConfig {
          return new SSLContextConfig(
             keystoreProvider, keystorePath, keystoreType, keystorePassword,
             truststoreProvider, truststorePath, truststoreType, truststorePassword,
-            crlPath, trustManagerFactoryPlugin, trustAll
+            crlPath, trustManagerFactoryPlugin, trustAll, keystoreAlias
          );
       }
 
@@ -119,6 +121,11 @@ public final class SSLContextConfig {
          return this;
       }
 
+      public Builder keystoreAlias(final String keystoreAlias) {
+         this.keystoreAlias = keystoreAlias;
+         return this;
+      }
+
       public Builder trustManagerFactoryPlugin(final String trustManagerFactoryPlugin) {
          this.trustManagerFactoryPlugin = trustManagerFactoryPlugin;
          return this;
@@ -140,6 +147,7 @@ public final class SSLContextConfig {
    private final String trustManagerFactoryPlugin;
    private final String crlPath;
    private final boolean trustAll;
+   private final String keystoreAlias;
    private final int hashCode;
 
    private SSLContextConfig(final String keystoreProvider,
@@ -152,7 +160,8 @@ public final class SSLContextConfig {
                             final String truststorePassword,
                             final String crlPath,
                             final String trustManagerFactoryPlugin,
-                            final boolean trustAll) {
+                            final boolean trustAll,
+                            final String keystoreAlias) {
       this.keystorePath = keystorePath;
       this.keystoreType = keystoreType;
       this.keystoreProvider = keystoreProvider;
@@ -164,10 +173,11 @@ public final class SSLContextConfig {
       this.trustManagerFactoryPlugin = trustManagerFactoryPlugin;
       this.crlPath = crlPath;
       this.trustAll = trustAll;
+      this.keystoreAlias = keystoreAlias;
       hashCode = Objects.hash(
          keystorePath, keystoreType, keystoreProvider,
          truststorePath, truststoreType, truststoreProvider,
-         crlPath, trustManagerFactoryPlugin, trustAll
+         crlPath, trustManagerFactoryPlugin, trustAll, keystoreAlias
       );
    }
 
@@ -186,7 +196,8 @@ public final class SSLContextConfig {
          && Objects.equals(truststoreProvider, other.truststoreProvider)
          && Objects.equals(crlPath, other.crlPath)
          && Objects.equals(trustManagerFactoryPlugin, other.trustManagerFactoryPlugin)
-         && trustAll == other.trustAll;
+         && trustAll == other.trustAll
+         && Objects.equals(keystoreAlias, other.keystoreAlias);
    }
 
    public String getCrlPath() {
@@ -238,6 +249,10 @@ public final class SSLContextConfig {
       return trustAll;
    }
 
+   public String getKeystoreAlias() {
+      return keystoreAlias;
+   }
+
    @Override
    public String toString() {
       return "SSLSupport [" +
@@ -252,6 +267,7 @@ public final class SSLContextConfig {
          ", crlPath=" + crlPath +
          ", trustAll=" + trustAll +
          ", trustManagerFactoryPlugin=" + trustManagerFactoryPlugin +
+         ", keystoreAlias=" + keystoreAlias +
          "]";
    }
 }

artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java

@@ -161,6 +161,8 @@ public class NettyAcceptor extends AbstractAcceptor {
 
    private final String keyStorePassword;
 
+   private final String keystoreAlias;
+
    private final String trustStoreProvider;
 
    private final String trustStoreType;
@@ -327,11 +329,14 @@ public class NettyAcceptor extends AbstractAcceptor {
 
          trustManagerFactoryPlugin = ConfigurationHelper.getStringProperty(TransportConstants.TRUST_MANAGER_FACTORY_PLUGIN_PROP_NAME, TransportConstants.DEFAULT_TRUST_MANAGER_FACTORY_PLUGIN, configuration);
 
+         keystoreAlias = ConfigurationHelper.getStringProperty(TransportConstants.KEYSTORE_ALIAS_PROP_NAME, TransportConstants.DEFAULT_KEYSTORE_ALIAS, configuration);
+
          sslContextConfig = SSLContextConfig.builder()
             .keystoreProvider(keyStoreProvider)
             .keystorePath(keyStorePath)
             .keystoreType(keyStoreType)
             .keystorePassword(keyStorePassword)
+            .keystoreAlias(keystoreAlias)
             .truststoreProvider(trustStoreProvider)
             .truststorePath(trustStorePath)
             .truststoreType(trustStoreType)
@@ -345,6 +350,7 @@ public class NettyAcceptor extends AbstractAcceptor {
          keyStoreType = TransportConstants.DEFAULT_KEYSTORE_TYPE;
          keyStorePath = TransportConstants.DEFAULT_KEYSTORE_PATH;
          keyStorePassword = TransportConstants.DEFAULT_KEYSTORE_PASSWORD;
+         keystoreAlias = TransportConstants.DEFAULT_KEYSTORE_ALIAS;
          trustStoreProvider = TransportConstants.DEFAULT_TRUSTSTORE_PROVIDER;
          trustStoreType = TransportConstants.DEFAULT_TRUSTSTORE_TYPE;
          trustStorePath = TransportConstants.DEFAULT_TRUSTSTORE_PATH;
@@ -563,12 +569,14 @@ public class NettyAcceptor extends AbstractAcceptor {
    }
 
    // only for testing purposes
-   public void setKeyStorePath(String keyStorePath) {
+   public void setKeyStoreParameters(String keyStorePath, String keyStoreAlias) {
       this.keyStorePath = keyStorePath;
       this.configuration.put(TransportConstants.KEYSTORE_PATH_PROP_NAME, keyStorePath);
+      this.configuration.put(TransportConstants.KEYSTORE_ALIAS_PROP_NAME, keyStoreAlias);
       sslContextConfig = SSLContextConfig.builder()
          .from(sslContextConfig)
          .keystorePath(keyStorePath)
+         .keystoreAlias(keyStoreAlias)
          .build();
    }
 

docs/user-manual/en/configuring-transports.md

@@ -379,6 +379,18 @@ additional properties:
   on the client is already making use of the standard Java system property.
   Default is `null`.
 
+- `keyStoreAlias`
+
+  When used on an `acceptor` this is the alias to select from the SSL key store
+  (specified via `keyStorePath`) to present to the client when it connects.
+
+  When used on a `connector` this is the alias to select from the SSL key store
+  (specified via `keyStorePath`) to present to the server when the client
+  connects to it. This is only relevant for a `connector` when using 2-way SSL
+  (i.e. mutual authentication).
+
+  Default is `null`.
+
 - `trustStorePath`
 
   When used on an `acceptor` this is the path to the server-side SSL key store

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/ssl/CoreClientOverOneWaySSLTest.java

@@ -61,25 +61,41 @@ import org.junit.runners.Parameterized;
  */
 @RunWith(value = Parameterized.class)
 public class CoreClientOverOneWaySSLTest extends ActiveMQTestBase {
-   String suffix = "";
 
-   @Parameterized.Parameters(name = "storeProvider={0}, storeType={1}")
+   public static final SimpleString QUEUE = new SimpleString("QueueOverSSL");
+
+   private boolean generateWarning;
+   private boolean useKeystoreAlias;
+   private String storeProvider;
+   private String storeType;
+   private String SERVER_SIDE_KEYSTORE;
+   private String CLIENT_SIDE_TRUSTSTORE;
+   private final String PASSWORD = "securepass";
+   private String suffix = "";
+
+   private ActiveMQServer server;
+
+   private TransportConfiguration tc;
+
+   @Parameterized.Parameters(name = "storeProvider={0}, storeType={1}, generateWarning={2}, useKeystoreAlias={3}")
    public static Collection getParameters() {
       return Arrays.asList(new Object[][]{
-         {TransportConstants.DEFAULT_KEYSTORE_PROVIDER, TransportConstants.DEFAULT_KEYSTORE_TYPE, false},
+         {TransportConstants.DEFAULT_KEYSTORE_PROVIDER, TransportConstants.DEFAULT_KEYSTORE_TYPE, false, false},
-         {"SunJCE", "JCEKS", false},
+         {TransportConstants.DEFAULT_KEYSTORE_PROVIDER, TransportConstants.DEFAULT_KEYSTORE_TYPE, false, true},
-         {"SUN", "JKS", false},
+         {"SunJCE", "JCEKS", false, false},
-         {"SunJSSE", "PKCS12", false},
+         {"SUN", "JKS", false, false},
-         {"JCEKS", null, true}, // for compatibility with old keyStoreProvider
+         {"SunJSSE", "PKCS12", false, false},
-         {"JKS", null, true},   // for compatibility with old keyStoreProvider
+         {"JCEKS", null, true, false}, // for compatibility with old keyStoreProvider
-         {"PKCS12", null, true} // for compatibility with old keyStoreProvider
+         {"JKS", null, true, false},   // for compatibility with old keyStoreProvider
+         {"PKCS12", null, true, false} // for compatibility with old keyStoreProvider
       });
    }
 
-   public CoreClientOverOneWaySSLTest(String storeProvider, String storeType, boolean generateWarning) {
+   public CoreClientOverOneWaySSLTest(String storeProvider, String storeType, boolean generateWarning, boolean useKeystoreAlias) {
       this.storeProvider = storeProvider;
       this.storeType = storeType;
       this.generateWarning = generateWarning;
+      this.useKeystoreAlias = useKeystoreAlias;
       suffix = storeType == null || storeType.length() == 0 ? storeProvider.toLowerCase() : storeType.toLowerCase();
       // keytool expects PKCS12 stores to use the extension "p12"
       if (suffix.equalsIgnoreCase("PKCS12")) {
@@ -89,19 +105,6 @@ public class CoreClientOverOneWaySSLTest extends ActiveMQTestBase {
       CLIENT_SIDE_TRUSTSTORE = "server-ca-truststore." + suffix;
    }
 
-   public static final SimpleString QUEUE = new SimpleString("QueueOverSSL");
-
-   private boolean generateWarning;
-   private String storeProvider;
-   private String storeType;
-   private String SERVER_SIDE_KEYSTORE;
-   private String CLIENT_SIDE_TRUSTSTORE;
-   private final String PASSWORD = "securepass";
-
-   private ActiveMQServer server;
-
-   private TransportConfiguration tc;
-
    @Before
    public void validateLogging() {
       AssertionLoggerHandler.startCapture();
@@ -522,7 +525,11 @@ public class CoreClientOverOneWaySSLTest extends ActiveMQTestBase {
 
       // reload the acceptor to reload the SSL stores
       NettyAcceptor acceptor = (NettyAcceptor) server.getRemotingService().getAcceptor("nettySSL");
-      acceptor.setKeyStorePath("other-" + SERVER_SIDE_KEYSTORE);
+      if (useKeystoreAlias) {
+         acceptor.setKeyStoreParameters("other-" + SERVER_SIDE_KEYSTORE, "other-server");
+      } else {
+         acceptor.setKeyStoreParameters("other-" + SERVER_SIDE_KEYSTORE, null);
+      }
       acceptor.reload();
 
       // create a session with the locator which failed previously proving that the SSL stores have been reloaded
@@ -990,6 +997,10 @@ public class CoreClientOverOneWaySSLTest extends ActiveMQTestBase {
          params.put(TransportConstants.KEYSTORE_PATH_PROP_NAME, SERVER_SIDE_KEYSTORE);
       }
       params.put(TransportConstants.KEYSTORE_PASSWORD_PROP_NAME, PASSWORD);
+      if (useKeystoreAlias) {
+         // the alias is specified when the keystore is created; see tests/security-resources/build.sh
+         params.put(TransportConstants.KEYSTORE_ALIAS_PROP_NAME, "server");
+      }
       params.put(TransportConstants.HOST_PROP_NAME, "localhost");
 
       if (cipherSuites != null) {

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/ssl/CoreClientOverTwoWaySSLTest.java

@@ -178,6 +178,74 @@ public class CoreClientOverTwoWaySSLTest extends ActiveMQTestBase {
       Assert.assertEquals(text, m.getBodyBuffer().readString());
    }
 
+   @Test
+   public void testTwoWaySSLChooseAliasPositive() throws Exception {
+      String text = RandomUtil.randomString();
+
+      tc.getParams().put(TransportConstants.SSL_ENABLED_PROP_NAME, true);
+      tc.getParams().put(TransportConstants.SSL_PROVIDER, clientSSLProvider);
+
+      tc.getParams().put(TransportConstants.KEYSTORE_PROVIDER_PROP_NAME, storeProvider);
+      tc.getParams().put(TransportConstants.KEYSTORE_TYPE_PROP_NAME, storeType);
+      tc.getParams().put(TransportConstants.KEYSTORE_PATH_PROP_NAME, CLIENT_SIDE_KEYSTORE);
+      tc.getParams().put(TransportConstants.KEYSTORE_PASSWORD_PROP_NAME, PASSWORD);
+
+      // the alias is specified when the keystore is created; see tests/security-resources/build.sh
+      tc.getParams().put(TransportConstants.KEYSTORE_ALIAS_PROP_NAME, "client");
+
+      tc.getParams().put(TransportConstants.TRUSTSTORE_PROVIDER_PROP_NAME, storeProvider);
+      tc.getParams().put(TransportConstants.TRUSTSTORE_TYPE_PROP_NAME, storeType);
+      tc.getParams().put(TransportConstants.TRUSTSTORE_PATH_PROP_NAME, CLIENT_SIDE_TRUSTSTORE);
+      tc.getParams().put(TransportConstants.TRUSTSTORE_PASSWORD_PROP_NAME, PASSWORD);
+
+      server.getRemotingService().addIncomingInterceptor(new MyInterceptor());
+
+      ServerLocator locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(tc));
+      ClientSessionFactory sf = createSessionFactory(locator);
+      ClientSession session = sf.createSession(false, true, true);
+      session.createQueue(new QueueConfiguration(CoreClientOverTwoWaySSLTest.QUEUE).setDurable(false));
+      ClientProducer producer = session.createProducer(CoreClientOverTwoWaySSLTest.QUEUE);
+
+      ClientMessage message = createTextMessage(session, text);
+      producer.send(message);
+
+      ClientConsumer consumer = session.createConsumer(CoreClientOverTwoWaySSLTest.QUEUE);
+      session.start();
+
+      ClientMessage m = consumer.receive(1000);
+      Assert.assertNotNull(m);
+      Assert.assertEquals(text, m.getBodyBuffer().readString());
+   }
+
+   @Test
+   public void testTwoWaySSLChooseAliasNegative() throws Exception {
+
+      tc.getParams().put(TransportConstants.SSL_ENABLED_PROP_NAME, true);
+      tc.getParams().put(TransportConstants.SSL_PROVIDER, clientSSLProvider);
+
+      tc.getParams().put(TransportConstants.KEYSTORE_PROVIDER_PROP_NAME, storeProvider);
+      tc.getParams().put(TransportConstants.KEYSTORE_TYPE_PROP_NAME, storeType);
+      tc.getParams().put(TransportConstants.KEYSTORE_PATH_PROP_NAME, CLIENT_SIDE_KEYSTORE);
+      tc.getParams().put(TransportConstants.KEYSTORE_PASSWORD_PROP_NAME, PASSWORD);
+      tc.getParams().put(TransportConstants.KEYSTORE_ALIAS_PROP_NAME, RandomUtil.randomString());
+
+      tc.getParams().put(TransportConstants.TRUSTSTORE_PROVIDER_PROP_NAME, storeProvider);
+      tc.getParams().put(TransportConstants.TRUSTSTORE_TYPE_PROP_NAME, storeType);
+      tc.getParams().put(TransportConstants.TRUSTSTORE_PATH_PROP_NAME, CLIENT_SIDE_TRUSTSTORE);
+      tc.getParams().put(TransportConstants.TRUSTSTORE_PASSWORD_PROP_NAME, PASSWORD);
+
+      server.getRemotingService().addIncomingInterceptor(new MyInterceptor());
+
+      ServerLocator locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(tc));
+      locator.setCallTimeout(500);
+      try {
+         ClientSessionFactory sf = createSessionFactory(locator);
+         fail();
+      } catch (ActiveMQNotConnectedException | ActiveMQConnectionTimedOutException e) {
+         // expected
+      }
+   }
+
    @Test
    public void testTwoWaySSLVerifyClientHost() throws Exception {
       NettyAcceptor acceptor = (NettyAcceptor) server.getRemotingService().getAcceptor("nettySSL");