[#2132] Migrate SSL support away from ThreadLocal, support key alias and default to pkcs12

activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioSslTransportFactory.java

@@ -22,6 +22,8 @@
 import java.net.URISyntaxException;
 import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
 
 import javax.net.ServerSocketFactory;
 import javax.net.SocketFactory;
@@ -34,14 +36,19 @@
 import org.apache.activemq.transport.tcp.TcpTransport;
 import org.apache.activemq.transport.tcp.TcpTransport.InitBuffer;
 import org.apache.activemq.transport.tcp.TcpTransportServer;
+import org.apache.activemq.util.IOExceptionSupport;
+import org.apache.activemq.util.IntrospectionSupport;
+import org.apache.activemq.util.URISupport;
 import org.apache.activemq.wireformat.WireFormat;
 
 public class AmqpNioSslTransportFactory extends AmqpNioTransportFactory {
 
-    protected SSLContext context;
-
     @Override
     protected TcpTransportServer createTcpTransportServer(URI location, ServerSocketFactory serverSocketFactory) throws IOException, URISyntaxException {
+        return createTcpTransportServer(location, serverSocketFactory, null);
+    }
+
+    protected TcpTransportServer createTcpTransportServer(URI location, ServerSocketFactory serverSocketFactory, final SSLContext context) throws IOException, URISyntaxException {
         return new TcpTransportServer(this, location, serverSocketFactory) {
             @Override
             protected Transport createTransport(Socket socket, WireFormat format) throws IOException {
@@ -73,13 +80,33 @@ public TcpTransport createTransport(WireFormat wireFormat, Socket socket,
 
     @Override
     public TransportServer doBind(URI location) throws IOException {
-        if (SslContext.getCurrentSslContext() != null) {
+        return doBind(location, null);
+    }
+
+    @Override
+    public TransportServer doBind(URI location, SslContext sslContext) throws IOException {
+        SSLContext context = null;
+        if (sslContext != null) {
             try {
-                context = SslContext.getCurrentSslContext().getSSLContext();
+                context = sslContext.getSSLContext();
             } catch (Exception e) {
                 throw new IOException(e);
             }
         }
-        return super.doBind(location);
+        try {
+            Map<String, String> options = new HashMap<String, String>(URISupport.parseParameters(location));
+
+            ServerSocketFactory serverSocketFactory = createServerSocketFactory();
+            TcpTransportServer server = createTcpTransportServer(location, serverSocketFactory, context);
+            server.setWireFormatFactory(createWireFormatFactory(options));
+            IntrospectionSupport.setProperties(server, options);
+            Map<String, Object> transportOptions = IntrospectionSupport.extractProperties(options, "transport.");
+            server.setTransportOption(transportOptions);
+            server.bind();
+
+            return server;
+        } catch (URISyntaxException e) {
+            throw IOExceptionSupport.create(e);
+        }
     }
 }

activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java

@@ -460,6 +460,7 @@ public boolean removeNetworkConnector(NetworkConnector connector) {
     public ProxyConnector addProxyConnector(ProxyConnector connector) throws Exception {
         URI uri = getVmConnectorURI();
         connector.setLocalUri(uri);
+        connector.setBrokerService(this);
         proxyConnectors.add(connector);
         if (isUseJmx()) {
             registerProxyConnectorMBean(connector);

activemq-broker/src/main/java/org/apache/activemq/broker/SslBrokerService.java

@@ -90,14 +90,8 @@ protected TransportServer createSslTransportServer(URI brokerURI, KeyManager[] k
             // If given an SSL URI, use an SSL TransportFactory and configure
             // it to use the given key and trust managers.
             SslTransportFactory transportFactory = new SslTransportFactory();
-
-            SslContext ctx = new SslContext(km, tm, random);
-            SslContext.setCurrentSslContext(ctx);
-            try {
-                return transportFactory.doBind(brokerURI);
-            } finally {
-                SslContext.setCurrentSslContext(null);
-            }
+            SslContext ctx = new DefaultSslContext(km, tm, random);
+            return transportFactory.doBind(brokerURI, ctx);
 
         } else {
             // Else, business as usual.

activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnector.java

@@ -22,6 +22,7 @@
 import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Optional;
 import java.util.StringTokenizer;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -82,6 +83,7 @@
     private boolean warnOnRemoteClose = false;
     private boolean displayStackTrace = false;
     private boolean autoStart = true;
+    private SslContext sslContext;
 
     LinkedList<String> peerBrokers = new LinkedList<String>();
     private AtomicBoolean started = new AtomicBoolean(false);
@@ -337,7 +339,8 @@
             throw new IllegalArgumentException(
                     "You must specify the brokerService property. Maybe this connector should be added to a broker?");
         }
-        return TransportFactorySupport.bind(brokerService, uri);
+        return TransportFactorySupport.bind(brokerService, uri,
+                Optional.ofNullable(sslContext).orElse(brokerService.getSslContext()));
     }
 
     public DiscoveryAgent getDiscoveryAgent() throws IOException {
@@ -604,7 +607,7 @@
 
     @Deprecated(forRemoval = true)
     @Override
     public int connectionCount() {
         return connections.size();
     }
 
@@ -704,6 +707,14 @@
         return autoStart;
     }
 
+    public SslContext getSslContext() {
+        return sslContext;
+    }
+
+    public void setSslContext(SslContext sslContext) {
+        this.sslContext = sslContext;
+    }
+
     @Override
     public int getConnectionCount() {
         return connections.size();

activemq-broker/src/main/java/org/apache/activemq/network/DiscoveryNetworkConnector.java

@@ -21,6 +21,7 @@
 import java.net.URISyntaxException;
 import java.util.Iterator;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
@@ -119,45 +120,37 @@ public void onServiceAdd(DiscoveryEvent event) {
 
             LOG.info("Establishing network connection from {} to {}", localURI, connectUri);
 
+            SslContext sslContext = Optional.ofNullable(getSslContext()).orElse(getBrokerService().getSslContext());
+
             Transport remoteTransport;
             Transport localTransport;
             try {
-                // Allows the transport to access the broker's ssl configuration.
-                if (getSslContext() != null) {
-                    SslContext.setCurrentSslContext(getSslContext());
-                } else {
-                    SslContext.setCurrentSslContext(getBrokerService().getSslContext());
-                }
+                remoteTransport = TransportFactory.connect(connectUri, sslContext);
+            } catch (Exception e) {
+                networkBridgeStatistics.getRemoteExceptionCount().increment();
+                LOG.warn("Could not connect to remote URI: {}: {}", connectUri, e.getMessage());
+                LOG.debug("Connection failure exception: ", e);
                 try {
-                    remoteTransport = TransportFactory.connect(connectUri);
-                } catch (Exception e) {
-                    networkBridgeStatistics.getRemoteExceptionCount().increment();
-                    LOG.warn("Could not connect to remote URI: {}: {}", connectUri, e.getMessage());
-                    LOG.debug("Connection failure exception: ", e);
-                    try {
-                        discoveryAgent.serviceFailed(event);
-                    } catch (IOException e1) {
-                        LOG.debug("Failure while handling create remote transport failure event: {}", e1.getMessage(), e1);
-                    }
-                    return;
+                    discoveryAgent.serviceFailed(event);
+                } catch (IOException e1) {
+                    LOG.debug("Failure while handling create remote transport failure event: {}", e1.getMessage(), e1);
                 }
-                try {
-                    localTransport = createLocalTransport();
-                } catch (Exception e) {
-                    networkBridgeStatistics.getLocalExceptionCount().increment();
-                    ServiceSupport.dispose(remoteTransport);
-                    LOG.warn("Could not connect to local URI: {}: {}", localURI, e.getMessage());
-                    LOG.debug("Connection failure exception: ", e);
+                return;
+            }
+            try {
+                localTransport = createLocalTransport();
+            } catch (Exception e) {
+                networkBridgeStatistics.getLocalExceptionCount().increment();
+                ServiceSupport.dispose(remoteTransport);
+                LOG.warn("Could not connect to local URI: {}: {}", localURI, e.getMessage());
+                LOG.debug("Connection failure exception: ", e);
 
-                    try {
-                        discoveryAgent.serviceFailed(event);
-                    } catch (IOException e1) {
-                        LOG.debug("Failure while handling create local transport failure event: {}", e1.getMessage(), e1);
-                    }
-                    return;
+                try {
+                    discoveryAgent.serviceFailed(event);
+                } catch (IOException e1) {
+                    LOG.debug("Failure while handling create local transport failure event: {}", e1.getMessage(), e1);
                 }
-            } finally {
-                SslContext.setCurrentSslContext(null);
+                return;
             }
             NetworkBridge bridge = createBridge(localTransport, remoteTransport, event);
             try {

activemq-broker/src/main/java/org/apache/activemq/network/MulticastNetworkConnector.java

@@ -17,6 +17,7 @@
 package org.apache.activemq.network;
 
 import java.net.URI;
+import java.util.Optional;
 
 import org.apache.activemq.transport.Transport;
 import org.apache.activemq.transport.TransportFactory;
@@ -103,7 +104,8 @@ protected void handleStart() throws Exception {
             if (remoteURI == null) {
                 throw new IllegalArgumentException("You must specify the remoteURI property");
             }
-            remoteTransport = TransportFactory.connect(remoteURI);
+            remoteTransport = TransportFactory.connect(remoteURI,
+                    Optional.ofNullable(getSslContext()).orElse(getBrokerService().getSslContext()));
         }
 
         if (localTransport == null) {

activemq-broker/src/main/java/org/apache/activemq/proxy/ProxyConnector.java

@@ -17,10 +17,13 @@
 package org.apache.activemq.proxy;
 
 import org.apache.activemq.Service;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.SslContext;
 import org.apache.activemq.transport.CompositeTransport;
 import org.apache.activemq.transport.Transport;
 import org.apache.activemq.transport.TransportAcceptListener;
 import org.apache.activemq.transport.TransportFactory;
+import org.apache.activemq.transport.TransportFactorySupport;
 import org.apache.activemq.transport.TransportFilter;
 import org.apache.activemq.transport.TransportServer;
 import org.apache.activemq.util.ServiceStopper;
@@ -31,6 +34,7 @@
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.Iterator;
+import java.util.Optional;
 import java.util.concurrent.CopyOnWriteArrayList;
 
 /**
@@ -44,6 +48,8 @@ public class ProxyConnector implements Service {
     private URI remote;
     private URI localUri;
     private String name;
+    private BrokerService brokerService;
+    private SslContext sslContext;
 
     /**
      * Should we proxy commands to the local broker using VM transport as well?
@@ -143,11 +149,11 @@ protected TransportServer createServer() throws IOException, URISyntaxException
         if (bind == null) {
             throw new IllegalArgumentException("You must specify either a server or the bind property");
         }
-        return TransportFactory.bind(bind);
+        return TransportFactorySupport.bind(brokerService, bind, resolveSslContext());
     }
 
     private Transport createRemoteTransport(final Transport local) throws Exception {
-        Transport transport = TransportFactory.compositeConnect(remote);
+        Transport transport = TransportFactory.compositeConnect(remote, resolveSslContext());
         CompositeTransport ct = transport.narrow(CompositeTransport.class);
         if (ct != null && localUri != null && proxyToLocalBroker) {
             ct.add(false, new URI[] { localUri });
@@ -190,6 +196,36 @@ public void setProxyToLocalBroker(boolean proxyToLocalBroker) {
         this.proxyToLocalBroker = proxyToLocalBroker;
     }
 
+    public BrokerService getBrokerService() {
+        return brokerService;
+    }
+
+    public void setBrokerService(BrokerService brokerService) {
+        this.brokerService = brokerService;
+    }
+
+    public SslContext getSslContext() {
+        return sslContext;
+    }
+
+    /**
+     * Sets the SSL context used when the bind or remote URI uses an SSL
+     * based transport. When not set, the broker's SSL context is used;
+     * when neither is set, the JVM default applies.
+     */
+    public void setSslContext(SslContext sslContext) {
+        this.sslContext = sslContext;
+    }
+
+    /**
+     * Resolves the SSL context at connect/bind time: the per-connector
+     * context wins, then the broker's context, then null (JVM default).
+     */
+    private SslContext resolveSslContext() {
+        return Optional.ofNullable(sslContext)
+                .orElse(brokerService != null ? brokerService.getSslContext() : null);
+    }
+
     protected Integer getConnectionCount() {
         return connections.size();
     }

activemq-broker/src/main/java/org/apache/activemq/transport/TransportFactorySupport.java

@@ -29,18 +29,15 @@
 public class TransportFactorySupport {
 
     public static TransportServer bind(BrokerService brokerService, URI location) throws IOException {
+        return bind(brokerService, location, brokerService != null ? brokerService.getSslContext() : null);
+    }
+
+    public static TransportServer bind(BrokerService brokerService, URI location, SslContext sslContext) throws IOException {
         TransportFactory tf = TransportFactory.findTransportFactory(location);
-        if( brokerService!=null && tf instanceof BrokerServiceAware) {
-            ((BrokerServiceAware)tf).setBrokerService(brokerService);
-        }
-        try {
-            if( brokerService!=null ) {
-                SslContext.setCurrentSslContext(brokerService.getSslContext());
-            }
-            return tf.doBind(location);
-        } finally {
-            SslContext.setCurrentSslContext(null);
+        if (brokerService != null && tf instanceof BrokerServiceAware) {
+            ((BrokerServiceAware) tf).setBrokerService(brokerService);
         }
+        return tf.doBind(location, sslContext);
     }
 
 }

activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoSslTransportFactory.java

@@ -29,6 +29,7 @@
 
 import org.apache.activemq.broker.BrokerService;
 import org.apache.activemq.broker.BrokerServiceAware;
+import org.apache.activemq.broker.SslContext;
 import org.apache.activemq.transport.TransportServer;
 import org.apache.activemq.transport.tcp.SslTransportFactory;
 import org.apache.activemq.transport.tcp.TcpTransport;
@@ -54,19 +55,30 @@ public void setBrokerService(BrokerService brokerService) {
 
     private Set<String> enabledProtocols;
 
-    /**
-     * Overriding to use SslTransportServer and allow for proper reflection.
-     */
     @Override
     public TransportServer doBind(final URI location) throws IOException {
+        return doBind(location, null);
+    }
+
+    @Override
+    public TransportServer doBind(final URI location, SslContext sslContext) throws IOException {
         try {
             Map<String, String> options = new HashMap<String, String>(URISupport.parseParameters(location));
 
             Map<String, Object> autoProperties = IntrospectionSupport.extractProperties(options, "auto.");
             this.enabledProtocols = AutoTransportUtils.parseProtocols((String) autoProperties.get("protocols"));
 
-            ServerSocketFactory serverSocketFactory = createServerSocketFactory();
-            AutoSslTransportServer server = createAutoSslTransportServer(location, (SSLServerSocketFactory)serverSocketFactory);
+            SSLServerSocketFactory serverSocketFactory;
+            if (sslContext != null) {
+                try {
+                    serverSocketFactory = sslContext.getSSLContext().getServerSocketFactory();
+                } catch (Exception e) {
+                    throw IOExceptionSupport.create(e);
+                }
+            } else {
+                serverSocketFactory = (SSLServerSocketFactory) SSLServerSocketFactory.getDefault();
+            }
+            AutoSslTransportServer server = createAutoSslTransportServer(location, serverSocketFactory);
             if (options.get("allowLinkStealing") != null){
                 allowLinkStealingSet = true;
             }