diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioSslTransportFactory.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioSslTransportFactory.java index 7858a561acd..c1a4e82f42f 100644 --- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioSslTransportFactory.java +++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioSslTransportFactory.java @@ -30,7 +30,6 @@ import org.apache.activemq.broker.SslContext; import org.apache.activemq.transport.Transport; -import org.apache.activemq.transport.TransportServer; import org.apache.activemq.transport.tcp.TcpTransport; import org.apache.activemq.transport.tcp.TcpTransport.InitBuffer; import org.apache.activemq.transport.tcp.TcpTransportServer; @@ -38,10 +37,14 @@ public class AmqpNioSslTransportFactory extends AmqpNioTransportFactory { - protected SSLContext context; - @Override protected TcpTransportServer createTcpTransportServer(URI location, ServerSocketFactory serverSocketFactory) throws IOException, URISyntaxException { + return createTcpTransportServer(location, serverSocketFactory, null); + } + + @Override + protected TcpTransportServer createTcpTransportServer(URI location, ServerSocketFactory serverSocketFactory, SslContext sslContext) throws IOException, URISyntaxException { + final SSLContext context = toSSLContext(sslContext); return new TcpTransportServer(this, location, serverSocketFactory) { @Override protected Transport createTransport(Socket socket, WireFormat format) throws IOException { @@ -70,16 +73,4 @@ public TcpTransport createTransport(WireFormat wireFormat, Socket socket, throws IOException { return new AmqpNioSslTransport(wireFormat, socket, engine, initBuffer, inputBuffer); } - - @Override - public TransportServer doBind(URI location) throws IOException { - if (SslContext.getCurrentSslContext() != null) { - try { - context = SslContext.getCurrentSslContext().getSSLContext(); - } catch (Exception e) { - throw new IOException(e); - } - } - return super.doBind(location); - } } diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioTransportFactory.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioTransportFactory.java index 4b6d9bd30f1..33a4af7c15d 100644 --- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioTransportFactory.java +++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpNioTransportFactory.java @@ -73,7 +73,7 @@ public TcpTransport createTransport(WireFormat wireFormat, Socket socket, @SuppressWarnings("rawtypes") @Override - public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) throws Exception { + public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) { transport = super.serverConfigure(transport, format, options); // strip off the mutex transport. diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpSslTransportFactory.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpSslTransportFactory.java index 00c72cafa4f..4ed675a3785 100644 --- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpSslTransportFactory.java +++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpSslTransportFactory.java @@ -54,7 +54,7 @@ public Transport compositeConfigure(Transport transport, WireFormat format, Map @SuppressWarnings("rawtypes") @Override - public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) throws Exception { + public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) { transport = super.serverConfigure(transport, format, options); // strip off the mutex transport. diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpTransportFactory.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpTransportFactory.java index 7f93b210ad1..13929187a40 100644 --- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpTransportFactory.java +++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpTransportFactory.java @@ -59,7 +59,7 @@ public void setBrokerService(BrokerService brokerService) { @SuppressWarnings("rawtypes") @Override - public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) throws Exception { + public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) { transport = super.serverConfigure(transport, format, options); // strip off the mutex transport. diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/SslConnectorOptionsAndContextTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/SslConnectorOptionsAndContextTest.java new file mode 100644 index 00000000000..7011e5a2236 --- /dev/null +++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/SslConnectorOptionsAndContextTest.java @@ -0,0 +1,289 @@ +/** + * 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.transport.amqp; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.net.URI; +import java.security.KeyStore; +import java.security.cert.X509Certificate; +import java.util.Arrays; +import java.util.Collection; + +import javax.net.ssl.KeyManager; +import javax.net.ssl.KeyManagerFactory; +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLSocket; +import javax.net.ssl.TrustManager; +import javax.net.ssl.TrustManagerFactory; +import javax.net.ssl.X509TrustManager; + +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.broker.DefaultSslContext; +import org.apache.activemq.broker.SslContext; +import org.apache.activemq.broker.TransportConnector; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Binds every SSL capable transport with URI options and a per-connector + * {@link SslContext}, then verifies through a raw TLS handshake what the + * accepted socket or engine actually negotiated. + * + *

Assertions are black-box on purpose: {@code transport.*} options are + * applied to the {@code SSLServerSocket} or the per-connection + * {@code SSLEngine}, not to the flags on the server object, so the server + * getters would not reflect them. The handshake completes before any protocol + * bytes flow, which is what lets one matrix cover OpenWire, AMQP, MQTT and + * STOMP transports alike. + */ +@Category(ParallelTest.class) +@RunWith(Parameterized.class) +public class SslConnectorOptionsAndContextTest { + + private static final Logger LOG = LoggerFactory.getLogger(SslConnectorOptionsAndContextTest.class); + + private static final char[] PASSWORD = "password".toCharArray(); + private static final String KEYSTORE = "keystore"; + private static final String KEYSTORE_ALIAS = "activemq"; + private static final String ALTERNATIVE_KEYSTORE = "alternative.keystore"; + private static final String ALTERNATIVE_ALIAS = "alternative"; + + private static final String HARDENED = "?transport.needClientAuth=true&transport.enabledProtocols=TLSv1.2"; + private static final String EXPECTED_PROTOCOL = "TLSv1.2"; + + @Parameterized.Parameters(name = "{0}") + public static Collection transports() { + return Arrays.asList(new Object[][] { + {"ssl"}, + {"nio+ssl"}, + {"auto+ssl"}, + {"auto+nio+ssl"}, + {"amqp+ssl"}, + {"amqp+nio+ssl"}, + {"mqtt+ssl"}, + {"mqtt+nio+ssl"}, + {"stomp+ssl"}, + {"stomp+nio+ssl"}, + }); + } + + @Parameterized.Parameter + public String transport; + + private BrokerService broker; + + @Before + public void setUp() { + broker = new BrokerService(); + broker.setPersistent(false); + broker.setUseJmx(false); + broker.setAdvisorySupport(false); + } + + @After + public void tearDown() throws Exception { + if (broker != null) { + broker.stop(); + broker.waitUntilStopped(); + } + } + + // ---- transport.* options reach the accepted socket / engine ---------- + + @Test(timeout = 60000) + public void testNeedClientAuthAndEnabledProtocolsApplied() throws Exception { + var connector = addConnector(HARDENED, sslContext(KEYSTORE)); + broker.start(); + broker.waitUntilStarted(); + + assertHardened(connector); + } + + /** + * Confirms current behavior: an unknown {@code transport.} option is silently + * dropped and the connector still binds with the valid options applied. + * If option validation is ever tightened this test must change with it. + */ + @Test(timeout = 60000) + public void testUnknownTransportOptionIsIgnored() throws Exception { + var connector = addConnector(HARDENED + "&transport.cheese=abc", sslContext(KEYSTORE)); + broker.start(); + broker.waitUntilStarted(); + + assertHardened(connector); + } + + /** + * Confirms current behavior: an unknown top level option is silently dropped + * on bind. The connect side rejects the same option with + * "Invalid connect parameters"; the bind side does not. + */ + @Test(timeout = 60000) + public void testUnknownTopLevelOptionIsIgnored() throws Exception { + var connector = addConnector(HARDENED + "&cheese=abc", sslContext(KEYSTORE)); + broker.start(); + broker.waitUntilStarted(); + + assertHardened(connector); + } + + /** + * Confirms current behavior: an unknown {@code wireFormat.} option is silently + * dropped on bind. + */ + @Test(timeout = 60000) + public void testUnknownWireFormatOptionIsIgnored() throws Exception { + var connector = addConnector(HARDENED + "&wireFormat.cheese=abc", sslContext(KEYSTORE)); + broker.start(); + broker.waitUntilStarted(); + + assertHardened(connector); + } + + /** + * Three connectors of the same transport on one broker: A and B carry + * their own SslContext with different identities, C has none and must + * fall back to the broker level context. Each must present the + * certificate of the context that applies to it. + */ + @Test(timeout = 60000) + public void testPerConnectorSslContextIsolation() throws Exception { + broker.setSslContext(sslContext(ALTERNATIVE_KEYSTORE)); + var connectorA = addConnector("", sslContext(KEYSTORE)); + var connectorB = addConnector("", sslContext(ALTERNATIVE_KEYSTORE)); + var connectorC = addConnector("", null); + broker.start(); + broker.waitUntilStarted(); + + var expectedA = certificate(KEYSTORE, KEYSTORE_ALIAS); + var expectedAlternative = certificate(ALTERNATIVE_KEYSTORE, ALTERNATIVE_ALIAS); + + assertEquals("connector A must present its own certificate", expectedA, handshake(connectorA, null).leaf()); + assertEquals("connector B must present its own certificate", expectedAlternative, handshake(connectorB, null).leaf()); + assertEquals("connector C must fall back to the broker certificate", expectedAlternative, handshake(connectorC, null).leaf()); + } + + /** + * Builds the connector by hand so the bind happens at broker start with the + * connector's own SslContext. {@code BrokerService.addConnector(URI)} binds + * eagerly with the broker level context at the time of the call, so a + * {@code setSslContext} on the connector it returns has no effect. + */ + private TransportConnector addConnector(String options, SslContext sslContext) throws Exception { + var connector = new TransportConnector(); + connector.setUri(new URI(transport + "://localhost:0" + options)); + connector.setSslContext(sslContext); + return broker.addConnector(connector); + } + + private void assertHardened(TransportConnector connector) throws Exception { + // A client presenting a certificate the connector trusts completes the + // handshake, and only TLSv1.2 can be negotiated. + var result = handshake(connector, keyManagers(KEYSTORE)); + LOG.info("{} negotiated {}", connector.getConnectUri(), result.protocol()); + assertEquals(EXPECTED_PROTOCOL, result.protocol()); + assertEquals(certificate(KEYSTORE, KEYSTORE_ALIAS), result.leaf()); + + // A client with no certificate is refused. Restricting to TLSv1.2 is + // what makes this deterministic: the server rejects the empty client + // Certificate message before its Finished, so the failure surfaces in + // startHandshake() rather than on a later read as it can with TLSv1.3. + var thrown = assertThrows(IOException.class, () -> handshake(connector, null)); + LOG.info("{} refused a client without a certificate: {}", connector.getConnectUri(), thrown.toString()); + } + + private static Handshake handshake(TransportConnector connector, KeyManager[] clientKeyManagers) throws Exception { + var catcher = new CertChainCatcher(); + var context = SSLContext.getInstance("TLS"); + context.init(clientKeyManagers, new TrustManager[] {catcher}, null); + + var uri = connector.getConnectUri(); + try (var socket = (SSLSocket) context.getSocketFactory().createSocket(uri.getHost(), uri.getPort())) { + socket.setSoTimeout(10000); + socket.startHandshake(); + assertNotNull("server did not present a certificate chain", catcher.serverCerts); + assertTrue("server presented an empty certificate chain", catcher.serverCerts.length > 0); + return new Handshake(socket.getSession().getProtocol(), catcher.serverCerts[0]); + } + } + + private record Handshake(String protocol, X509Certificate leaf) { + } + + private static SslContext sslContext(String keystoreName) throws Exception { + var keyStore = loadKeyStore(keystoreName); + var kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); + kmf.init(keyStore, PASSWORD); + var tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); + tmf.init(keyStore); + return new DefaultSslContext(kmf.getKeyManagers(), tmf.getTrustManagers(), null); + } + + private static KeyManager[] keyManagers(String keystoreName) throws Exception { + var kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); + kmf.init(loadKeyStore(keystoreName), PASSWORD); + return kmf.getKeyManagers(); + } + + private static X509Certificate certificate(String keystoreName, String alias) throws Exception { + var certificate = (X509Certificate) loadKeyStore(keystoreName).getCertificate(alias); + assertNotNull("alias " + alias + " not found in " + keystoreName, certificate); + return certificate; + } + + private static KeyStore loadKeyStore(String keystoreName) throws Exception { + var url = SslConnectorOptionsAndContextTest.class.getClassLoader().getResource(keystoreName); + assertNotNull("test keystore not on classpath: " + keystoreName, url); + var keyStore = KeyStore.getInstance("jks"); + try (var in = new FileInputStream(new File(url.toURI()))) { + keyStore.load(in, PASSWORD); + } + return keyStore; + } + + /** Accepts any server certificate and records the chain it presented. */ + private static final class CertChainCatcher implements X509TrustManager { + volatile X509Certificate[] serverCerts; + + @Override + public void checkClientTrusted(X509Certificate[] chain, String authType) { + } + + @Override + public void checkServerTrusted(X509Certificate[] chain, String authType) { + serverCerts = chain; + } + + @Override + public X509Certificate[] getAcceptedIssuers() { + return new X509Certificate[0]; + } + } +} diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java b/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java index 1ffd8d79133..e4ac456b2ec 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java +++ b/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java @@ -329,7 +329,9 @@ private String getBrokerVersion() { } /** - * Adds a new transport connector for the given bind address + * Adds a new transport connector for the given bind address. + * Binds immediately using the broker-wide sslContext; for a per-connector + * sslContext build a TransportConnector and use {@link #addConnector(TransportConnector)}. * * @return the newly created and added transport connector */ @@ -338,7 +340,9 @@ public TransportConnector addConnector(String bindAddress) throws Exception { } /** - * Adds a new transport connector for the given bind address + * Adds a new transport connector for the given bind address. + * Binds immediately using the broker-wide sslContext; for a per-connector + * sslContext build a TransportConnector and use {@link #addConnector(TransportConnector)}. * * @return the newly created and added transport connector */ @@ -460,6 +464,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); @@ -2453,7 +2458,7 @@ protected DestinationInterceptor[] createDefaultDestinationInterceptor() { /** * Strategy method to add interceptors to the broker * - * @throws IOException + * @throws Exception */ protected Broker addInterceptors(Broker broker) throws Exception { if (isAdvisorySupport()) { diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/SslBrokerService.java b/activemq-broker/src/main/java/org/apache/activemq/broker/SslBrokerService.java index 4ef23abf35e..b5522407417 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/broker/SslBrokerService.java +++ b/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. diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnector.java b/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnector.java index b3fd69f61ca..938165195c2 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnector.java +++ b/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 @@ public class TransportConnector implements Connector, BrokerServiceAware { private boolean warnOnRemoteClose = false; private boolean displayStackTrace = false; private boolean autoStart = true; + private SslContext sslContext; LinkedList peerBrokers = new LinkedList(); private AtomicBoolean started = new AtomicBoolean(false); @@ -337,7 +339,8 @@ protected TransportServer createTransportServer() throws IOException, URISyntaxE 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 { @@ -704,6 +707,14 @@ public boolean isAutoStart() { return autoStart; } + public SslContext getSslContext() { + return sslContext; + } + + public void setSslContext(SslContext sslContext) { + this.sslContext = sslContext; + } + @Override public int getConnectionCount() { return connections.size(); diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/DiscoveryNetworkConnector.java b/activemq-broker/src/main/java/org/apache/activemq/network/DiscoveryNetworkConnector.java index 5532f3f98a5..1059d1f3636 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/network/DiscoveryNetworkConnector.java +++ b/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 { diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/MulticastNetworkConnector.java b/activemq-broker/src/main/java/org/apache/activemq/network/MulticastNetworkConnector.java index f3b581ea240..a104528591d 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/network/MulticastNetworkConnector.java +++ b/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) { diff --git a/activemq-broker/src/main/java/org/apache/activemq/proxy/ProxyConnector.java b/activemq-broker/src/main/java/org/apache/activemq/proxy/ProxyConnector.java index b36faafc252..d569ed36e55 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/proxy/ProxyConnector.java +++ b/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(); } diff --git a/activemq-broker/src/main/java/org/apache/activemq/transport/TransportFactorySupport.java b/activemq-broker/src/main/java/org/apache/activemq/transport/TransportFactorySupport.java index afc20ce932b..9f223334e8b 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/transport/TransportFactorySupport.java +++ b/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); } } diff --git a/activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoSslTransportFactory.java b/activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoSslTransportFactory.java index d1ad524c9ba..ad45258af2b 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoSslTransportFactory.java +++ b/activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoSslTransportFactory.java @@ -24,11 +24,11 @@ import java.util.Map; import java.util.Set; -import javax.net.ServerSocketFactory; import javax.net.ssl.SSLServerSocketFactory; 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 +54,16 @@ public void setBrokerService(BrokerService brokerService) { private Set enabledProtocols; - /** - * Overriding to use SslTransportServer and allow for proper reflection. - */ @Override - public TransportServer doBind(final URI location) throws IOException { + public TransportServer doBind(final URI location, SslContext sslContext) throws IOException { try { Map options = new HashMap(URISupport.parseParameters(location)); Map autoProperties = IntrospectionSupport.extractProperties(options, "auto."); this.enabledProtocols = AutoTransportUtils.parseProtocols((String) autoProperties.get("protocols")); - ServerSocketFactory serverSocketFactory = createServerSocketFactory(); - AutoSslTransportServer server = createAutoSslTransportServer(location, (SSLServerSocketFactory)serverSocketFactory); + SSLServerSocketFactory serverSocketFactory = (SSLServerSocketFactory) createServerSocketFactory(sslContext); + AutoSslTransportServer server = createAutoSslTransportServer(location, serverSocketFactory); if (options.get("allowLinkStealing") != null){ allowLinkStealingSet = true; } diff --git a/activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoTcpTransportFactory.java b/activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoTcpTransportFactory.java index 10ddca0f3de..4fce71e2d6b 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoTcpTransportFactory.java +++ b/activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoTcpTransportFactory.java @@ -28,6 +28,7 @@ import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.BrokerServiceAware; +import org.apache.activemq.broker.SslContext; import org.apache.activemq.openwire.OpenWireFormatFactory; import org.apache.activemq.transport.TransportServer; import org.apache.activemq.transport.tcp.TcpTransport; @@ -56,7 +57,7 @@ public void setBrokerService(BrokerService brokerService) { @Override - public TransportServer doBind(final URI location) throws IOException { + public TransportServer doBind(final URI location, SslContext sslContext) throws IOException { try { Map options = new HashMap(URISupport.parseParameters(location)); diff --git a/activemq-broker/src/main/java/org/apache/activemq/transport/auto/nio/AutoNioSslTransportFactory.java b/activemq-broker/src/main/java/org/apache/activemq/transport/auto/nio/AutoNioSslTransportFactory.java index 8a29ab246dc..47b753d0f03 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/transport/auto/nio/AutoNioSslTransportFactory.java +++ b/activemq-broker/src/main/java/org/apache/activemq/transport/auto/nio/AutoNioSslTransportFactory.java @@ -26,6 +26,7 @@ import java.util.Set; import javax.net.ServerSocketFactory; +import javax.net.ssl.SSLContext; import javax.net.ssl.SSLEngine; import org.apache.activemq.broker.BrokerService; @@ -62,6 +63,12 @@ public void setBrokerService(BrokerService brokerService) { @Override protected AutoNIOSSLTransportServer createTcpTransportServer(URI location, ServerSocketFactory serverSocketFactory) throws IOException, URISyntaxException { + return createTcpTransportServer(location, serverSocketFactory, null); + } + + @Override + protected AutoNIOSSLTransportServer createTcpTransportServer(URI location, ServerSocketFactory serverSocketFactory, SslContext sslContext) throws IOException, URISyntaxException { + final SSLContext context = toSSLContext(sslContext); return new AutoNIOSSLTransportServer(context, this, location, serverSocketFactory, brokerService, enabledProtocols) { @Override @@ -94,23 +101,15 @@ protected Transport createTransport(Socket socket, WireFormat format, SSLEngine private Set enabledProtocols; @Override - public TransportServer doBind(final URI location) throws IOException { + public TransportServer doBind(final URI location, SslContext sslContext) throws IOException { try { - if (SslContext.getCurrentSslContext() != null) { - try { - context = SslContext.getCurrentSslContext().getSSLContext(); - } catch (Exception e) { - throw new IOException(e); - } - } - Map options = new HashMap(URISupport.parseParameters(location)); Map autoProperties = IntrospectionSupport.extractProperties(options, "auto."); this.enabledProtocols = AutoTransportUtils.parseProtocols((String) autoProperties.get("protocols")); - ServerSocketFactory serverSocketFactory = createServerSocketFactory(); - AutoTcpTransportServer server = createTcpTransportServer(location, serverSocketFactory); + ServerSocketFactory serverSocketFactory = createServerSocketFactory(sslContext); + AutoTcpTransportServer server = createTcpTransportServer(location, serverSocketFactory, sslContext); server.setWireFormatFactory(new OpenWireFormatFactory()); if (options.get("allowLinkStealing") != null){ allowLinkStealingSet = true; diff --git a/activemq-broker/src/main/java/org/apache/activemq/transport/auto/nio/AutoNioTransportFactory.java b/activemq-broker/src/main/java/org/apache/activemq/transport/auto/nio/AutoNioTransportFactory.java index ec9b2787a53..f3165255f1c 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/transport/auto/nio/AutoNioTransportFactory.java +++ b/activemq-broker/src/main/java/org/apache/activemq/transport/auto/nio/AutoNioTransportFactory.java @@ -28,6 +28,7 @@ import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.BrokerServiceAware; +import org.apache.activemq.broker.SslContext; import org.apache.activemq.openwire.OpenWireFormatFactory; import org.apache.activemq.transport.TransportServer; import org.apache.activemq.transport.auto.AutoTcpTransportServer; @@ -84,7 +85,7 @@ protected TcpTransport createTransport(Socket socket, WireFormat format, TcpTran private Set enabledProtocols; @Override - public TransportServer doBind(final URI location) throws IOException { + public TransportServer doBind(final URI location, SslContext sslContext) throws IOException { try { Map options = new HashMap(URISupport.parseParameters(location)); diff --git a/activemq-broker/src/main/java/org/apache/activemq/transport/peer/PeerTransportFactory.java b/activemq-broker/src/main/java/org/apache/activemq/transport/peer/PeerTransportFactory.java index c5b6d0c07b1..70b3a3a5793 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/transport/peer/PeerTransportFactory.java +++ b/activemq-broker/src/main/java/org/apache/activemq/transport/peer/PeerTransportFactory.java @@ -44,23 +44,24 @@ public class PeerTransportFactory extends TransportFactory { private static final IdGenerator ID_GENERATOR = new IdGenerator("peer-"); @Override - public Transport doConnect(URI location) throws Exception { + public Transport doConnect(URI location) throws IOException, URISyntaxException { VMTransportFactory vmTransportFactory = createTransportFactory(location); return vmTransportFactory.doConnect(location); } @Override - public Transport doCompositeConnect(URI location) throws Exception { + public Transport doCompositeConnect(URI location) throws IOException, URISyntaxException { VMTransportFactory vmTransportFactory = createTransportFactory(location); return vmTransportFactory.doCompositeConnect(location); } /** * @param location - * @return the converted URI + * @return a VMTransportFactory that connects to the vm:// location derived from the peer URI + * @throws IOException * @throws URISyntaxException */ - private VMTransportFactory createTransportFactory(URI location) throws IOException { + private VMTransportFactory createTransportFactory(URI location) throws IOException, URISyntaxException { try { String group = location.getHost(); String broker = URISupport.stripPrefix(location.getPath(), "/"); @@ -82,12 +83,12 @@ private VMTransportFactory createTransportFactory(URI location) throws IOExcepti final String finalGroup = group; VMTransportFactory rc = new VMTransportFactory() { @Override - public Transport doConnect(URI ignore) throws Exception { + public Transport doConnect(URI ignore) throws IOException, URISyntaxException { return super.doConnect(finalLocation); }; @Override - public Transport doCompositeConnect(URI ignore) throws Exception { + public Transport doCompositeConnect(URI ignore) throws IOException, URISyntaxException { return super.doCompositeConnect(finalLocation); }; }; diff --git a/activemq-broker/src/main/java/org/apache/activemq/transport/vm/VMTransportFactory.java b/activemq-broker/src/main/java/org/apache/activemq/transport/vm/VMTransportFactory.java index 63b94dac5d9..f72027c0dea 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/transport/vm/VMTransportFactory.java +++ b/activemq-broker/src/main/java/org/apache/activemq/transport/vm/VMTransportFactory.java @@ -72,12 +72,12 @@ public VMTransportFactory() { } @Override - public Transport doConnect(URI location) throws Exception { + public Transport doConnect(URI location) throws IOException, URISyntaxException { return VMTransportServer.configure(doCompositeConnect(location)); } @Override - public Transport doCompositeConnect(URI location) throws Exception { + public Transport doCompositeConnect(URI location) throws IOException, URISyntaxException { URI brokerURI; String host; Map options; @@ -134,11 +134,17 @@ public Transport doCompositeConnect(URI location) throws Exception { // cause multiple brokers to be started. synchronized (BrokerRegistry.getInstance().getRegistryMutext()) { broker = lookupBroker(BrokerRegistry.getInstance(), host, waitForStart); - if (broker == null) { - if (!create) { - throw new IOException("Broker named '" + host + "' does not exist."); - } - try { + // Broker create/start and connector.start() below are the only + // calls here that throw a bare checked Exception; wrap those as + // IOException to satisfy the narrowed throws clause. IOException + // and any RuntimeException (e.g. the schema-guard + // IllegalArgumentException, or invalid broker config) are + // re-thrown unchanged so their type/contract is preserved. + try { + if (broker == null) { + if (!create) { + throw new IOException("Broker named '" + host + "' does not exist."); + } validateBrokerCreationSchema(host, brokerURI); if (brokerFactoryHandler != null) { broker = brokerFactoryHandler.createBroker(brokerURI); @@ -147,22 +153,24 @@ public Transport doCompositeConnect(URI location) throws Exception { } broker.start(); MDC.put("activemq.broker", broker.getBrokerName()); - } catch (URISyntaxException e) { - throw IOExceptionSupport.create(e); + BROKERS.put(host, broker); + BrokerRegistry.getInstance().getRegistryMutext().notifyAll(); } - BROKERS.put(host, broker); - BrokerRegistry.getInstance().getRegistryMutext().notifyAll(); - } - server = SERVERS.get(host); - if (server == null) { - server = (VMTransportServer)bind(location, true); - TransportConnector connector = new TransportConnector(server); - connector.setBrokerService(broker); - connector.setUri(location); - connector.setTaskRunnerFactory(broker.getTaskRunnerFactory()); - connector.start(); - CONNECTORS.put(host, connector); + server = SERVERS.get(host); + if (server == null) { + server = (VMTransportServer)bind(location, true); + TransportConnector connector = new TransportConnector(server); + connector.setBrokerService(broker); + connector.setUri(location); + connector.setTaskRunnerFactory(broker.getTaskRunnerFactory()); + connector.start(); + CONNECTORS.put(host, connector); + } + } catch (IOException | RuntimeException e) { + throw e; + } catch (Exception e) { + throw IOExceptionSupport.create(e); } } diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQSslConnectionFactory.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQSslConnectionFactory.java index 69e159e0c3a..3bb510a6da9 100644 --- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQSslConnectionFactory.java +++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQSslConnectionFactory.java @@ -36,7 +36,9 @@ import javax.net.ssl.TrustManagerFactory; import org.apache.activemq.broker.SslContext; +import org.apache.activemq.broker.DefaultSslContext; import org.apache.activemq.transport.Transport; +import org.apache.activemq.transport.TransportFactory; import org.apache.activemq.util.JMSExceptionSupport; /** @@ -108,20 +110,31 @@ public void setKeyAndTrustManagers(final KeyManager[] km, final TrustManager[] t */ @Override protected Transport createTransport() throws JMSException { - SslContext existing = SslContext.getCurrentSslContext(); try { if (keyStore != null || trustStore != null) { keyManager = createKeyManager(); trustManager = createTrustManager(); } if (keyManager != null || trustManager != null) { - SslContext.setCurrentSslContext(new SslContext(keyManager, trustManager, secureRandom)); + SslContext sslContext = new DefaultSslContext(keyManager, trustManager, secureRandom); + URI connectBrokerUL = brokerURL; + String scheme = brokerURL.getScheme(); + if (scheme != null) { + if (scheme.equals("auto")) { + connectBrokerUL = new URI(brokerURL.toString().replace("auto", "tcp")); + } else if (scheme.equals("auto+ssl")) { + connectBrokerUL = new URI(brokerURL.toString().replace("auto+ssl", "ssl")); + } else if (scheme.equals("auto+nio")) { + connectBrokerUL = new URI(brokerURL.toString().replace("auto+nio", "nio")); + } else if (scheme.equals("auto+nio+ssl")) { + connectBrokerUL = new URI(brokerURL.toString().replace("auto+nio+ssl", "nio+ssl")); + } + } + return TransportFactory.connect(connectBrokerUL, sslContext); } return super.createTransport(); } catch (Exception e) { throw JMSExceptionSupport.create("Could not create Transport. Reason: " + e, e); - } finally { - SslContext.setCurrentSslContext(existing); } } diff --git a/activemq-client/src/main/java/org/apache/activemq/broker/CompatibleSslContext.java b/activemq-client/src/main/java/org/apache/activemq/broker/CompatibleSslContext.java new file mode 100644 index 00000000000..2c3693ece99 --- /dev/null +++ b/activemq-client/src/main/java/org/apache/activemq/broker/CompatibleSslContext.java @@ -0,0 +1,155 @@ +/** + * 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.broker; + +import java.security.KeyManagementException; +import java.security.NoSuchAlgorithmException; +import java.security.NoSuchProviderException; +import java.security.SecureRandom; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import javax.net.ssl.KeyManager; +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManager; + +/** + * An {@link SslContext} that is API-compatible with {@link ThreadLocalSslContext} + * but carries no ThreadLocal state. + * + *

This class exposes the same protected fields, list-based getters/setters, + * and mutator methods that {@link ThreadLocalSslContext} does, so subclasses + * (such as {@code SpringSslContext}) can be reparented onto this class without + * any source or binary breakage. + * + *

The SSLContext is lazily created on first call to {@link #getSSLContext()} + * and can be replaced directly via {@link #setSSLContext(SSLContext)}. + */ +public class CompatibleSslContext extends SslContext { + + protected String protocol = "TLS"; + protected String provider = null; + protected List keyManagers = new ArrayList<>(); + protected List trustManagers = new ArrayList<>(); + protected SecureRandom secureRandom; + + private volatile boolean initialized; + private volatile SSLContext sslContext; + + public CompatibleSslContext() { + } + + public CompatibleSslContext(KeyManager[] km, TrustManager[] tm, SecureRandom random) { + if (km != null) { + setKeyManagers(Arrays.asList(km)); + } + if (tm != null) { + setTrustManagers(Arrays.asList(tm)); + } + setSecureRandom(random); + } + + public KeyManager[] getKeyManagersAsArray() { + KeyManager[] rc = new KeyManager[keyManagers.size()]; + return keyManagers.toArray(rc); + } + + public TrustManager[] getTrustManagersAsArray() { + TrustManager[] rc = new TrustManager[trustManagers.size()]; + return trustManagers.toArray(rc); + } + + public void addKeyManager(KeyManager km) { + keyManagers.add(km); + } + + public boolean removeKeyManager(KeyManager km) { + return keyManagers.remove(km); + } + + public void addTrustManager(TrustManager tm) { + trustManagers.add(tm); + } + + public boolean removeTrustManager(TrustManager tm) { + return trustManagers.remove(tm); + } + + public List getKeyManagers() { + return keyManagers; + } + + public void setKeyManagers(List keyManagers) { + this.keyManagers = keyManagers; + } + + public List getTrustManagers() { + return trustManagers; + } + + public void setTrustManagers(List trustManagers) { + this.trustManagers = trustManagers; + } + + public SecureRandom getSecureRandom() { + return secureRandom; + } + + public void setSecureRandom(SecureRandom secureRandom) { + this.secureRandom = secureRandom; + } + + public String getProtocol() { + return protocol; + } + + public void setProtocol(String protocol) { + this.protocol = protocol; + } + + public String getProvider() { + return provider; + } + + public void setProvider(String provider) { + this.provider = provider; + } + + @Override + public SSLContext getSSLContext() throws NoSuchProviderException, NoSuchAlgorithmException, KeyManagementException { + if (!initialized) { + synchronized (this) { + if (!initialized) { + if (provider == null) { + sslContext = SSLContext.getInstance(protocol); + } else { + sslContext = SSLContext.getInstance(protocol, provider); + } + sslContext.init(getKeyManagersAsArray(), getTrustManagersAsArray(), getSecureRandom()); + initialized = true; + } + } + } + return sslContext; + } + + public synchronized void setSSLContext(SSLContext sslContext) { + this.sslContext = sslContext; + initialized = true; + } +} diff --git a/activemq-client/src/main/java/org/apache/activemq/broker/DefaultSslContext.java b/activemq-client/src/main/java/org/apache/activemq/broker/DefaultSslContext.java new file mode 100644 index 00000000000..3c67bb83843 --- /dev/null +++ b/activemq-client/src/main/java/org/apache/activemq/broker/DefaultSslContext.java @@ -0,0 +1,116 @@ +/** + * 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.broker; + +import java.security.KeyManagementException; +import java.security.NoSuchAlgorithmException; +import java.security.NoSuchProviderException; +import java.security.SecureRandom; + +import javax.net.ssl.KeyManager; +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManager; + +/** + * A simple {@link SslContext} that holds key/trust managers directly and + * creates a single {@link SSLContext} on first use. + * + *

This is the recommended replacement for {@link ThreadLocalSslContext} + * when explicit parameter passing is used throughout the transport chain. + * It carries no ThreadLocal state and does not support runtime certificate + * reload. + */ +public class DefaultSslContext extends SslContext { + + private String protocol = "TLS"; + private String provider; + private KeyManager[] keyManagers; + private TrustManager[] trustManagers; + private SecureRandom secureRandom; + + private volatile SSLContext sslContext; + + public DefaultSslContext() { + } + + public DefaultSslContext(KeyManager[] km, TrustManager[] tm, SecureRandom random) { + this.keyManagers = km; + this.trustManagers = tm; + this.secureRandom = random; + } + + @Override + public SSLContext getSSLContext() throws NoSuchProviderException, NoSuchAlgorithmException, KeyManagementException { + if (sslContext == null) { + synchronized (this) { + if (sslContext == null) { + SSLContext ctx; + if (provider == null) { + ctx = SSLContext.getInstance(protocol); + } else { + ctx = SSLContext.getInstance(protocol, provider); + } + ctx.init(keyManagers, trustManagers, secureRandom); + sslContext = ctx; + } + } + } + return sslContext; + } + + // --- Bean properties --- + + public String getProtocol() { + return protocol; + } + + public void setProtocol(String protocol) { + this.protocol = protocol; + } + + public String getProvider() { + return provider; + } + + public void setProvider(String provider) { + this.provider = provider; + } + + public KeyManager[] getKeyManagers() { + return keyManagers; + } + + public void setKeyManagers(KeyManager[] keyManagers) { + this.keyManagers = keyManagers; + } + + public TrustManager[] getTrustManagers() { + return trustManagers; + } + + public void setTrustManagers(TrustManager[] trustManagers) { + this.trustManagers = trustManagers; + } + + public SecureRandom getSecureRandom() { + return secureRandom; + } + + public void setSecureRandom(SecureRandom secureRandom) { + this.secureRandom = secureRandom; + } +} diff --git a/activemq-client/src/main/java/org/apache/activemq/broker/SslContext.java b/activemq-client/src/main/java/org/apache/activemq/broker/SslContext.java index 61e534a5e9e..bb34153995e 100644 --- a/activemq-client/src/main/java/org/apache/activemq/broker/SslContext.java +++ b/activemq-client/src/main/java/org/apache/activemq/broker/SslContext.java @@ -19,124 +19,46 @@ import java.security.KeyManagementException; import java.security.NoSuchAlgorithmException; import java.security.NoSuchProviderException; -import java.security.SecureRandom; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; -import javax.net.ssl.KeyManager; import javax.net.ssl.SSLContext; -import javax.net.ssl.TrustManager; /** - * A holder of SSL configuration. + * Provides an {@link SSLContext} for SSL/TLS transport connectors and + * network connectors. + * + *

Implementations may be static (see {@link ThreadLocalSslContext}) or + * support runtime certificate reload. */ -public class SslContext { - - protected String protocol = "TLS"; - protected String provider = null; - protected List keyManagers = new ArrayList(); - protected List trustManagers = new ArrayList(); - protected SecureRandom secureRandom; - private volatile boolean initialized; - private SSLContext sslContext; - - private static final ThreadLocal current = new ThreadLocal(); - - public SslContext() { - } - - public SslContext(KeyManager[] km, TrustManager[] tm, SecureRandom random) { - if( km!=null ) { - setKeyManagers(Arrays.asList(km)); - } - if( tm!=null ) { - setTrustManagers(Arrays.asList(tm)); - } - setSecureRandom(random); - } - - static public void setCurrentSslContext(SslContext bs) { - current.set(bs); - } - static public SslContext getCurrentSslContext() { - return current.get(); - } - - public KeyManager[] getKeyManagersAsArray() { - KeyManager rc[] = new KeyManager[keyManagers.size()]; - return keyManagers.toArray(rc); - } - public TrustManager[] getTrustManagersAsArray() { - TrustManager rc[] = new TrustManager[trustManagers.size()]; - return trustManagers.toArray(rc); - } - - public void addKeyManager(KeyManager km) { - keyManagers.add(km); - } - public boolean removeKeyManager(KeyManager km) { - return keyManagers.remove(km); - } - public void addTrustManager(TrustManager tm) { - trustManagers.add(tm); - } - public boolean removeTrustManager(TrustManager tm) { - return trustManagers.remove(tm); - } - - public List getKeyManagers() { - return keyManagers; - } - public void setKeyManagers(List keyManagers) { - this.keyManagers = keyManagers; - } - public List getTrustManagers() { - return trustManagers; - } - public void setTrustManagers(List trustManagers) { - this.trustManagers = trustManagers; - } - public SecureRandom getSecureRandom() { - return secureRandom; - } - public void setSecureRandom(SecureRandom secureRandom) { - this.secureRandom = secureRandom; - } - - public String getProtocol() { - return protocol; - } - public void setProtocol(String protocol) { - this.protocol = protocol; - } - public String getProvider() { - return provider; - } - public void setProvider(String provider) { - this.provider = provider; +public abstract class SslContext { + + /** + * Returns a fully initialised {@link SSLContext} ready for use by + * transport factories. + */ + public abstract SSLContext getSSLContext() throws NoSuchProviderException, NoSuchAlgorithmException, KeyManagementException; + + /** + * Reload certificates from the underlying key/trust material. + * The default implementation is a no-op; reloadable implementations + * override this to swap in new credentials without restarting the + * broker. + */ + public void reload() throws Exception { } - public SSLContext getSSLContext() throws NoSuchProviderException, NoSuchAlgorithmException, KeyManagementException { - if (!initialized) { - synchronized (this) { - if (!initialized) { - if (provider == null) { - sslContext = SSLContext.getInstance(protocol); - } else { - sslContext = SSLContext.getInstance(protocol, provider); - } - sslContext.init(getKeyManagersAsArray(), getTrustManagersAsArray(), getSecureRandom()); - initialized = true; - } - } - } - return sslContext; + /** + * @deprecated Use explicit parameter passing instead of ThreadLocal propagation. + */ + @Deprecated + public static void setCurrentSslContext(SslContext ctx) { + ThreadLocalSslContext.setCurrent(ctx); } - public synchronized void setSSLContext(SSLContext sslContext) { - this.sslContext = sslContext; - initialized = true; + + /** + * @deprecated Use explicit parameter passing instead of ThreadLocal propagation. + */ + @Deprecated + public static SslContext getCurrentSslContext() { + return ThreadLocalSslContext.getCurrent(); } - - } diff --git a/activemq-client/src/main/java/org/apache/activemq/broker/ThreadLocalSslContext.java b/activemq-client/src/main/java/org/apache/activemq/broker/ThreadLocalSslContext.java new file mode 100644 index 00000000000..f2c948e133c --- /dev/null +++ b/activemq-client/src/main/java/org/apache/activemq/broker/ThreadLocalSslContext.java @@ -0,0 +1,162 @@ +/** + * 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.broker; + +import java.security.KeyManagementException; +import java.security.NoSuchAlgorithmException; +import java.security.NoSuchProviderException; +import java.security.SecureRandom; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import javax.net.ssl.KeyManager; +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManager; + +/** + * The original {@link SslContext} implementation that uses a {@link ThreadLocal} + * to propagate SSL configuration through the transport factory call chain. + * + *

This class preserves full backward compatibility with the pre-interface + * {@code SslContext} class. New code should prefer explicit parameter passing + * and, where runtime certificate reload is needed, use a + * {@code ReloadableSslContext} instead. + */ +public class ThreadLocalSslContext extends SslContext { + + private static final ThreadLocal current = new ThreadLocal<>(); + + protected String protocol = "TLS"; + protected String provider = null; + protected List keyManagers = new ArrayList<>(); + protected List trustManagers = new ArrayList<>(); + protected SecureRandom secureRandom; + private volatile boolean initialized; + private SSLContext sslContext; + + public ThreadLocalSslContext() { + } + + public ThreadLocalSslContext(KeyManager[] km, TrustManager[] tm, SecureRandom random) { + if (km != null) { + setKeyManagers(Arrays.asList(km)); + } + if (tm != null) { + setTrustManagers(Arrays.asList(tm)); + } + setSecureRandom(random); + } + + static void setCurrent(SslContext ctx) { + current.set(ctx); + } + + static SslContext getCurrent() { + return current.get(); + } + + public KeyManager[] getKeyManagersAsArray() { + KeyManager[] rc = new KeyManager[keyManagers.size()]; + return keyManagers.toArray(rc); + } + + public TrustManager[] getTrustManagersAsArray() { + TrustManager[] rc = new TrustManager[trustManagers.size()]; + return trustManagers.toArray(rc); + } + + public void addKeyManager(KeyManager km) { + keyManagers.add(km); + } + + public boolean removeKeyManager(KeyManager km) { + return keyManagers.remove(km); + } + + public void addTrustManager(TrustManager tm) { + trustManagers.add(tm); + } + + public boolean removeTrustManager(TrustManager tm) { + return trustManagers.remove(tm); + } + + public List getKeyManagers() { + return keyManagers; + } + + public void setKeyManagers(List keyManagers) { + this.keyManagers = keyManagers; + } + + public List getTrustManagers() { + return trustManagers; + } + + public void setTrustManagers(List trustManagers) { + this.trustManagers = trustManagers; + } + + public SecureRandom getSecureRandom() { + return secureRandom; + } + + public void setSecureRandom(SecureRandom secureRandom) { + this.secureRandom = secureRandom; + } + + public String getProtocol() { + return protocol; + } + + public void setProtocol(String protocol) { + this.protocol = protocol; + } + + public String getProvider() { + return provider; + } + + public void setProvider(String provider) { + this.provider = provider; + } + + @Override + public SSLContext getSSLContext() throws NoSuchProviderException, NoSuchAlgorithmException, KeyManagementException { + if (!initialized) { + synchronized (this) { + if (!initialized) { + if (provider == null) { + sslContext = SSLContext.getInstance(protocol); + } else { + sslContext = SSLContext.getInstance(protocol, provider); + } + sslContext.init(getKeyManagersAsArray(), getTrustManagersAsArray(), getSecureRandom()); + initialized = true; + } + } + } + return sslContext; + } + + public synchronized void setSSLContext(SSLContext sslContext) { + this.sslContext = sslContext; + initialized = true; + } +} + diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/TransportFactory.java b/activemq-client/src/main/java/org/apache/activemq/transport/TransportFactory.java index 184a5d63911..4fe58d66e6f 100644 --- a/activemq-client/src/main/java/org/apache/activemq/transport/TransportFactory.java +++ b/activemq-client/src/main/java/org/apache/activemq/transport/TransportFactory.java @@ -17,16 +17,15 @@ package org.apache.activemq.transport; import java.io.IOException; -import java.net.MalformedURLException; import java.net.URI; import java.net.URISyntaxException; -import java.net.UnknownHostException; import java.util.HashMap; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.Executor; +import org.apache.activemq.broker.SslContext; import org.apache.activemq.util.FactoryFinder; import org.apache.activemq.util.IOExceptionSupport; import org.apache.activemq.util.IntrospectionSupport; @@ -49,11 +48,38 @@ public abstract class TransportFactory { public abstract TransportServer doBind(URI location) throws IOException; - public Transport doConnect(URI location, Executor ex) throws Exception { + /** + * Default implementation delegates to the single-arg method, ignoring the + * SslContext. Subclasses (e.g. SslTransportFactory) override to use the + * context for SSL socket creation. + */ + public TransportServer doBind(URI location, SslContext sslContext) throws IOException { + return doBind(location); + } + + /** + * Default implementation delegates to the single-arg method, ignoring the + * SslContext. Subclasses (e.g. SslTransportFactory) override to use the + * context for SSL socket creation. + */ + public Transport doConnect(URI location, SslContext sslContext) throws IOException, URISyntaxException { + return doConnect(location); + } + + /** + * Default implementation delegates to the single-arg method, ignoring the + * SslContext. Subclasses (e.g. SslTransportFactory) override to use the + * context for SSL socket creation. + */ + public Transport doCompositeConnect(URI location, SslContext sslContext) throws IOException, URISyntaxException { + return doCompositeConnect(location); + } + + public Transport doConnect(URI location, Executor ex) throws IOException, URISyntaxException { return doConnect(location); } - public Transport doCompositeConnect(URI location, Executor ex) throws Exception { + public Transport doCompositeConnect(URI location, Executor ex) throws IOException, URISyntaxException { return doCompositeConnect(location); } @@ -62,22 +88,29 @@ public Transport doCompositeConnect(URI location, Executor ex) throws Exception * * @param location * @return the transport - * @throws Exception + * @throws IOException + * @throws URISyntaxException */ - public static Transport connect(URI location) throws Exception { + public static Transport connect(URI location) throws IOException, URISyntaxException { TransportFactory tf = findTransportFactory(location); return tf.doConnect(location); } + public static Transport connect(URI location, SslContext sslContext) throws IOException, URISyntaxException { + TransportFactory tf = findTransportFactory(location); + return tf.doConnect(location, sslContext); + } + /** * Creates a normal transport. * * @param location * @param ex * @return the transport - * @throws Exception + * @throws IOException + * @throws URISyntaxException */ - public static Transport connect(URI location, Executor ex) throws Exception { + public static Transport connect(URI location, Executor ex) throws IOException, URISyntaxException { TransportFactory tf = findTransportFactory(location); return tf.doConnect(location, ex); } @@ -88,13 +121,19 @@ public static Transport connect(URI location, Executor ex) throws Exception { * * @param location * @return the Transport - * @throws Exception + * @throws IOException + * @throws URISyntaxException */ - public static Transport compositeConnect(URI location) throws Exception { + public static Transport compositeConnect(URI location) throws IOException, URISyntaxException { TransportFactory tf = findTransportFactory(location); return tf.doCompositeConnect(location); } + public static Transport compositeConnect(URI location, SslContext sslContext) throws IOException, URISyntaxException { + TransportFactory tf = findTransportFactory(location); + return tf.doCompositeConnect(location, sslContext); + } + /** * Creates a slimmed down transport that is more efficient so that it can be * used by composite transports like reliable and HA. @@ -102,9 +141,10 @@ public static Transport compositeConnect(URI location) throws Exception { * @param location * @param ex * @return the Transport - * @throws Exception + * @throws IOException + * @throws URISyntaxException */ - public static Transport compositeConnect(URI location, Executor ex) throws Exception { + public static Transport compositeConnect(URI location, Executor ex) throws IOException, URISyntaxException { TransportFactory tf = findTransportFactory(location); return tf.doCompositeConnect(location, ex); } @@ -114,33 +154,39 @@ public static TransportServer bind(URI location) throws IOException { return tf.doBind(location); } - public Transport doConnect(URI location) throws Exception { - try { - Map options = new HashMap(URISupport.parseParameters(location)); - if( !options.containsKey("wireFormat.host") ) { - options.put("wireFormat.host", location.getHost()); - } - WireFormat wf = createWireFormat(options); - Transport transport = createTransport(location, wf); - Transport rc = configure(transport, wf, options); - //remove auto - IntrospectionSupport.extractProperties(options, "auto."); + public Transport doConnect(URI location) throws IOException, URISyntaxException { + return doConnectInternal(location, null, false); + } - if (!options.isEmpty()) { - throw new IllegalArgumentException("Invalid connect parameters: " + options); - } - return rc; - } catch (URISyntaxException e) { - throw IOExceptionSupport.create(e); - } + public Transport doCompositeConnect(URI location) throws IOException, URISyntaxException { + return doConnectInternal(location, null, true); } - public Transport doCompositeConnect(URI location) throws Exception { + /** + * Shared implementation behind {@link #doConnect(URI)} / + * {@link #doCompositeConnect(URI)} and the SslContext-carrying overrides in + * SSL capable subclasses. {@code composite} selects the slimmed-down form + * used by reliable/HA transports (no {@code wireFormat.host} default, + * {@link #compositeConfigure} instead of {@link #configure}, and no + * {@code auto.*} strip). The SslContext is threaded to + * {@link #createTransport(URI, WireFormat, SslContext)} — plain transports + * ignore it, SSL capable ones derive their socket factory from it — so the + * connect template lives here once rather than being copied per transport. + */ + protected Transport doConnectInternal(URI location, SslContext sslContext, boolean composite) throws IOException { try { Map options = new HashMap(URISupport.parseParameters(location)); + if (!composite && !options.containsKey("wireFormat.host")) { + options.put("wireFormat.host", location.getHost()); + } WireFormat wf = createWireFormat(options); - Transport transport = createTransport(location, wf); - Transport rc = compositeConfigure(transport, wf, options); + Transport transport = createTransport(location, wf, sslContext); + Transport rc = composite ? compositeConfigure(transport, wf, options) + : configure(transport, wf, options); + if (!composite) { + //remove auto + IntrospectionSupport.extractProperties(options, "auto."); + } if (!options.isEmpty()) { throw new IllegalArgumentException("Invalid connect parameters: " + options); } @@ -163,12 +209,25 @@ public static void registerTransportFactory(String scheme, TransportFactory tf) * Factory method to create a new transport * * @throws IOException - * @throws UnknownHostException */ - protected Transport createTransport(URI location, WireFormat wf) throws MalformedURLException, UnknownHostException, IOException { + protected Transport createTransport(URI location, WireFormat wf) throws IOException { throw new IOException("createTransport() method not implemented!"); } + /** + * SSL-aware createTransport override point used by {@link #doConnectInternal}. + * The default ignores the SslContext and delegates to + * {@link #createTransport(URI, WireFormat)}; + * SSL capable subclasses (e.g. TcpTransportFactory) override this to derive + * their socket factory from the context, so the connect template does not + * have to be duplicated per transport. + * + * @param sslContext the SslContext to use, or null for the JVM default. + */ + protected Transport createTransport(URI location, WireFormat wf, SslContext sslContext) throws IOException { + return createTransport(location, wf); + } + /** * @param location * @return @@ -225,10 +284,10 @@ protected String getDefaultWireFormatType() { * @param wf * @param options * @return - * @throws Exception + * @throws IOException */ @SuppressWarnings("rawtypes") - public Transport configure(Transport transport, WireFormat wf, Map options) throws Exception { + public Transport configure(Transport transport, WireFormat wf, Map options) throws IOException { transport = compositeConfigure(transport, wf, options); transport = new MutexTransport(transport); @@ -247,10 +306,9 @@ public Transport configure(Transport transport, WireFormat wf, Map options) thro * @param format * @param options * @return - * @throws Exception */ @SuppressWarnings("rawtypes") - public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) throws Exception { + public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) { if (options.containsKey(THREAD_NAME_FILTER)) { transport = new ThreadNameFilter(transport); } diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/discovery/DiscoveryTransportFactory.java b/activemq-client/src/main/java/org/apache/activemq/transport/discovery/DiscoveryTransportFactory.java index 6a0def814ff..11f9cce44ef 100644 --- a/activemq-client/src/main/java/org/apache/activemq/transport/discovery/DiscoveryTransportFactory.java +++ b/activemq-client/src/main/java/org/apache/activemq/transport/discovery/DiscoveryTransportFactory.java @@ -21,6 +21,7 @@ import java.util.HashMap; import java.util.Map; +import org.apache.activemq.broker.SslContext; import org.apache.activemq.transport.CompositeTransport; import org.apache.activemq.transport.Transport; import org.apache.activemq.transport.TransportServer; @@ -33,10 +34,15 @@ * */ public class DiscoveryTransportFactory extends FailoverTransportFactory { - + public Transport createTransport(CompositeData compositeData) throws IOException { - Map parameters = new HashMap(compositeData.getParameters()); - FailoverTransport failoverTransport = createTransport(parameters); + return createTransport(compositeData, null); + } + + @Override + public Transport createTransport(CompositeData compositeData, SslContext sslContext) throws IOException { + Map parameters = new HashMap<>(compositeData.getParameters()); + FailoverTransport failoverTransport = createTransport(parameters, sslContext); return createTransport(failoverTransport, compositeData, parameters); } diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java b/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java index 379e8aa3711..5be8cf3b5ce 100644 --- a/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java +++ b/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java @@ -119,7 +119,7 @@ public class FailoverTransport implements CompositeTransport { private boolean updateURIsSupported = true; private boolean reconnectSupported = true; // remember for reconnect thread - private SslContext brokerSslContext; + private SslContext sslContext; private String updateURIsURL = null; private boolean rebalanceUpdateURIs = true; private boolean doRebalance = false; @@ -132,8 +132,13 @@ public class FailoverTransport implements CompositeTransport { private String nestedExtraQueryOptions; private volatile boolean shuttingDown = false; + @SuppressWarnings("deprecation") public FailoverTransport() { - brokerSslContext = SslContext.getCurrentSslContext(); + this(null); + } + + public FailoverTransport(SslContext sslContext) { + this.sslContext = sslContext; stateTracker.setTrackTransactions(true); // Setup a task that is used to reconnect the a connection async. reconnectTaskFactory = new TaskRunnerFactory(); @@ -1016,13 +1021,11 @@ final boolean doReconnect() { while ((transport != null || iter.hasNext()) && (connectedTransport.get() == null && !disposed)) { try { - SslContext.setCurrentSslContext(brokerSslContext); - // We could be starting with a backup and if so we wait to grab a // URI from the pool until next time around. if (transport == null) { uri = addExtraQueryOptions(iter.next()); - transport = TransportFactory.compositeConnect(uri); + transport = TransportFactory.compositeConnect(uri, this.sslContext); } LOG.debug("Attempting {}th connect to: {}", connectFailures, uri); @@ -1081,7 +1084,6 @@ final boolean doReconnect() { } } } finally { - SslContext.setCurrentSslContext(null); } } } @@ -1199,11 +1201,10 @@ final boolean buildBackups() { URI uri = addExtraQueryOptions(iter.next()); if (connectedTransportURI != null && !connectedTransportURI.equals(uri)) { try { - SslContext.setCurrentSslContext(brokerSslContext); BackupTransport bt = new BackupTransport(this); bt.setUri(uri); if (!backups.contains(bt)) { - Transport t = TransportFactory.compositeConnect(uri); + Transport t = TransportFactory.compositeConnect(uri, this.sslContext); t.setTransportListener(bt); t.start(); bt.setTransport(t); @@ -1227,8 +1228,6 @@ final boolean buildBackups() { } } catch (Exception e) { LOG.debug("Failed to build backup ", e); - } finally { - SslContext.setCurrentSslContext(null); } } } diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransportFactory.java b/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransportFactory.java index 14b5b16e70b..6dceea6307e 100644 --- a/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransportFactory.java +++ b/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransportFactory.java @@ -20,6 +20,8 @@ import java.net.URI; import java.net.URISyntaxException; import java.util.Map; + +import org.apache.activemq.broker.SslContext; import org.apache.activemq.transport.MutexTransport; import org.apache.activemq.transport.ResponseCorrelator; import org.apache.activemq.transport.Transport; @@ -33,8 +35,13 @@ public class FailoverTransportFactory extends TransportFactory { @Override public Transport doConnect(URI location) throws IOException { + return doConnect(location, (SslContext) null); + } + + @Override + public Transport doConnect(URI location, SslContext sslContext) throws IOException { try { - Transport transport = createTransport(URISupport.parseComposite(location)); + Transport transport = createTransport(URISupport.parseComposite(location), sslContext); transport = new MutexTransport(transport); transport = new ResponseCorrelator(transport); return transport; @@ -52,14 +59,22 @@ public Transport doCompositeConnect(URI location) throws IOException { } } - /** - * @param compositData - * @return - * @throws IOException - */ + @Override + public Transport doCompositeConnect(URI location, SslContext sslContext) throws IOException { + try { + return createTransport(URISupport.parseComposite(location), sslContext); + } catch (URISyntaxException e) { + throw new IOException("Invalid location: " + location); + } + } + public Transport createTransport(CompositeData compositData) throws IOException { + return createTransport(compositData, null); + } + + public Transport createTransport(CompositeData compositData, SslContext sslContext) throws IOException { Map options = compositData.getParameters(); - FailoverTransport transport = createTransport(options); + FailoverTransport transport = createTransport(options, sslContext); if (!options.isEmpty()) { throw new IllegalArgumentException("Invalid connect parameters: " + options); } @@ -68,7 +83,11 @@ public Transport createTransport(CompositeData compositData) throws IOException } public FailoverTransport createTransport(Map parameters) throws IOException { - FailoverTransport transport = new FailoverTransport(); + return createTransport(parameters, null); + } + + public FailoverTransport createTransport(Map parameters, SslContext sslContext) throws IOException { + FailoverTransport transport = new FailoverTransport(sslContext); Map nestedExtraQueryOptions = IntrospectionSupport.extractProperties(parameters, "nested."); IntrospectionSupport.setProperties(transport, parameters); try { diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java b/activemq-client/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java index d2e2f30cd45..6080eedb0ed 100644 --- a/activemq-client/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java +++ b/activemq-client/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java @@ -26,6 +26,7 @@ import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicInteger; +import org.apache.activemq.broker.SslContext; import org.apache.activemq.command.Command; import org.apache.activemq.command.ConsumerInfo; import org.apache.activemq.command.Message; @@ -69,6 +70,7 @@ public class FanoutTransport implements CompositeTransport { private boolean started; private final ArrayList transports = new ArrayList(); + private final SslContext sslContext; private int connectedCount; private int minAckCount = 2; @@ -158,6 +160,11 @@ public void onException(IOException error) { } public FanoutTransport() { + this(null); + } + + public FanoutTransport(SslContext sslContext) { + this.sslContext = sslContext; // Setup a task that is used to reconnect the a connection async. reconnectTaskFactory = new TaskRunnerFactory(); reconnectTaskFactory.init(); @@ -212,7 +219,7 @@ private boolean doConnect() { try { LOG.debug("Stopped: " + this); LOG.debug("Attempting connect to: " + uri); - Transport t = TransportFactory.compositeConnect(uri); + Transport t = TransportFactory.compositeConnect(uri, sslContext); fanoutHandler.transport = t; t.setTransportListener(fanoutHandler); if (started) { diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/fanout/FanoutTransportFactory.java b/activemq-client/src/main/java/org/apache/activemq/transport/fanout/FanoutTransportFactory.java index e35842e3d87..9e3563bf2af 100644 --- a/activemq-client/src/main/java/org/apache/activemq/transport/fanout/FanoutTransportFactory.java +++ b/activemq-client/src/main/java/org/apache/activemq/transport/fanout/FanoutTransportFactory.java @@ -21,6 +21,7 @@ import java.net.URISyntaxException; import java.util.Map; +import org.apache.activemq.broker.SslContext; import org.apache.activemq.transport.MutexTransport; import org.apache.activemq.transport.ResponseCorrelator; import org.apache.activemq.transport.Transport; @@ -35,8 +36,17 @@ public class FanoutTransportFactory extends TransportFactory { public Transport doConnect(URI location) throws IOException { + return doConnectInternal(location, null); + } + + @Override + public Transport doConnect(URI location, SslContext sslContext) throws IOException { + return doConnectInternal(location, sslContext); + } + + private Transport doConnectInternal(URI location, SslContext sslContext) throws IOException { try { - Transport transport = createTransport(location); + Transport transport = createTransport(location, sslContext); transport = new MutexTransport(transport); transport = new ResponseCorrelator(transport); return transport; @@ -47,7 +57,16 @@ public Transport doConnect(URI location) throws IOException { public Transport doCompositeConnect(URI location) throws IOException { try { - return createTransport(location); + return createTransport(location, (SslContext) null); + } catch (URISyntaxException e) { + throw new IOException("Invalid location: " + location); + } + } + + @Override + public Transport doCompositeConnect(URI location, SslContext sslContext) throws IOException { + try { + return createTransport(location, sslContext); } catch (URISyntaxException e) { throw new IOException("Invalid location: " + location); } @@ -60,15 +79,23 @@ public Transport doCompositeConnect(URI location) throws IOException { * @throws URISyntaxException */ public Transport createTransport(URI location) throws IOException, URISyntaxException { + return createTransport(location, (SslContext) null); + } + + public Transport createTransport(URI location, SslContext sslContext) throws IOException, URISyntaxException { CompositeData compositeData = URISupport.parseComposite(location); Map parameters = compositeData.getParameters(); - FanoutTransport fanoutTransport = createTransport(parameters); + FanoutTransport fanoutTransport = createTransport(parameters, sslContext); DiscoveryTransport discoveryTransport = DiscoveryTransportFactory.createTransport(fanoutTransport, compositeData, parameters); return discoveryTransport; } public FanoutTransport createTransport(Map parameters) throws IOException { - FanoutTransport transport = new FanoutTransport(); + return createTransport(parameters, null); + } + + public FanoutTransport createTransport(Map parameters, SslContext sslContext) throws IOException { + FanoutTransport transport = new FanoutTransport(sslContext); IntrospectionSupport.setProperties(transport, parameters); return transport; } diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/mock/MockTransportFactory.java b/activemq-client/src/main/java/org/apache/activemq/transport/mock/MockTransportFactory.java index 9a4b37671fd..522d467b1bf 100644 --- a/activemq-client/src/main/java/org/apache/activemq/transport/mock/MockTransportFactory.java +++ b/activemq-client/src/main/java/org/apache/activemq/transport/mock/MockTransportFactory.java @@ -32,7 +32,7 @@ public class MockTransportFactory extends TransportFactory { @Override - public Transport doConnect(URI location) throws URISyntaxException, Exception { + public Transport doConnect(URI location) throws IOException, URISyntaxException { Transport transport = createTransport(URISupport.parseComposite(location)); transport = new MutexTransport(transport); transport = new ResponseCorrelator(transport); @@ -40,16 +40,17 @@ public Transport doConnect(URI location) throws URISyntaxException, Exception { } @Override - public Transport doCompositeConnect(URI location) throws URISyntaxException, Exception { + public Transport doCompositeConnect(URI location) throws IOException, URISyntaxException { return createTransport(URISupport.parseComposite(location)); } /** * @param compositData * @return a new Transport instance. - * @throws Exception + * @throws IOException + * @throws URISyntaxException */ - public Transport createTransport(CompositeData compositData) throws Exception { + public Transport createTransport(CompositeData compositData) throws IOException, URISyntaxException { MockTransport transport = new MockTransport(TransportFactory.compositeConnect(compositData.getComponents()[0])); IntrospectionSupport.setProperties(transport, compositData.getParameters()); return transport; diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransportFactory.java b/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransportFactory.java index 405314f617e..45c7c3ea5dc 100644 --- a/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransportFactory.java +++ b/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransportFactory.java @@ -33,7 +33,6 @@ import org.apache.activemq.broker.SslContext; import org.apache.activemq.transport.Transport; -import org.apache.activemq.transport.TransportServer; import org.apache.activemq.transport.tcp.SslTransport; import org.apache.activemq.transport.tcp.TcpTransport; import org.apache.activemq.transport.tcp.TcpTransport.InitBuffer; @@ -41,29 +40,22 @@ import org.apache.activemq.util.IOExceptionSupport; import org.apache.activemq.util.IntrospectionSupport; import org.apache.activemq.wireformat.WireFormat; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; public class NIOSSLTransportFactory extends NIOTransportFactory { - private static final Logger LOG = LoggerFactory.getLogger(NIOSSLTransportFactory.class); - - protected SSLContext context; - @Override protected TcpTransportServer createTcpTransportServer(URI location, ServerSocketFactory serverSocketFactory) throws IOException, URISyntaxException { - return new NIOSSLTransportServer(context, this, location, serverSocketFactory); + return createTcpTransportServer(location, serverSocketFactory, null); } + /** + * Overriding to create an NIO SSL transport server that uses the given + * SslContext for accepted connections. + * + * @param sslContext the SslContext to use, or null for the JVM default. + */ @Override - public TransportServer doBind(URI location) throws IOException { - if (SslContext.getCurrentSslContext() != null) { - try { - context = SslContext.getCurrentSslContext().getSSLContext(); - } catch (Exception e) { - throw new IOException(e); - } - } - return super.doBind(location); + protected TcpTransportServer createTcpTransportServer(URI location, ServerSocketFactory serverSocketFactory, SslContext sslContext) throws IOException, URISyntaxException { + return new NIOSSLTransportServer(toSSLContext(sslContext), this, location, serverSocketFactory); } /** @@ -84,25 +76,20 @@ public Transport compositeConfigure(Transport transport, WireFormat format, Map } /** - * Overriding to use SslTransports. + * Overriding to derive the SSL socket factory from the given SslContext, + * falling back to the JVM default when none is supplied. */ @Override - protected Transport createTransport(URI location, WireFormat wf) throws UnknownHostException, IOException { + protected SocketFactory createSocketFactory(SslContext sslContext) throws IOException { + SSLContext context = toSSLContext(sslContext); + return context != null ? context.getSocketFactory() : createSocketFactory(); + } - URI localLocation = null; - String path = location.getPath(); - // see if the path is a local URI location - if (path != null && path.length() > 0) { - int localPortIndex = path.indexOf(':'); - try { - Integer.parseInt(path.substring(localPortIndex + 1, path.length())); - String localString = location.getScheme() + ":/" + path; - localLocation = new URI(localString); - } catch (Exception e) { - LOG.warn("path isn't a valid local location for SslTransport to use", e); - } - } - SocketFactory socketFactory = createSocketFactory(); + /** + * Overriding to use SslTransports. + */ + @Override + protected TcpTransport createTcpTransport(WireFormat wf, SocketFactory socketFactory, URI location, URI localLocation) throws UnknownHostException, IOException { return new SslTransport(wf, (SSLSocketFactory) socketFactory, location, localLocation, false); } @@ -113,26 +100,9 @@ public TcpTransport createTransport(WireFormat wireFormat, Socket socket, return new NIOSSLTransport(wireFormat, socket, engine, initBuffer, inputBuffer); } - /** - * Creates a new SSL SocketFactory. The given factory will use user-provided - * key and trust managers (if the user provided them). - * - * @return Newly created (Ssl)SocketFactory. - * @throws IOException - */ @Override protected SocketFactory createSocketFactory() throws IOException { - if (SslContext.getCurrentSslContext() != null) { - SslContext ctx = SslContext.getCurrentSslContext(); - try { - return ctx.getSSLContext().getSocketFactory(); - } catch (Exception e) { - throw IOExceptionSupport.create(e); - } - } else { - return SSLSocketFactory.getDefault(); - } - + return SSLSocketFactory.getDefault(); } } diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/tcp/SslTransportFactory.java b/activemq-client/src/main/java/org/apache/activemq/transport/tcp/SslTransportFactory.java index b289eec25cf..ff599097db3 100644 --- a/activemq-client/src/main/java/org/apache/activemq/transport/tcp/SslTransportFactory.java +++ b/activemq-client/src/main/java/org/apache/activemq/transport/tcp/SslTransportFactory.java @@ -21,25 +21,21 @@ import java.net.URI; import java.net.URISyntaxException; import java.net.UnknownHostException; -import java.util.HashMap; import java.util.Map; import javax.net.ServerSocketFactory; import javax.net.SocketFactory; +import javax.net.ssl.SSLContext; import javax.net.ssl.SSLServerSocketFactory; import javax.net.ssl.SSLSocket; import javax.net.ssl.SSLSocketFactory; import org.apache.activemq.broker.SslContext; import org.apache.activemq.transport.Transport; -import org.apache.activemq.transport.TransportServer; import org.apache.activemq.transport.tcp.TcpTransport.InitBuffer; import org.apache.activemq.util.IOExceptionSupport; import org.apache.activemq.util.IntrospectionSupport; -import org.apache.activemq.util.URISupport; import org.apache.activemq.wireformat.WireFormat; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * An implementation of the TcpTransportFactory using SSL. The major @@ -49,28 +45,22 @@ */ public class SslTransportFactory extends TcpTransportFactory { - private static final Logger LOG = LoggerFactory.getLogger(SslTransportFactory.class); + /** + * Overriding to derive the SSL server socket factory from the given + * SslContext, falling back to the JVM default when none is supplied. + */ + @Override + protected ServerSocketFactory createServerSocketFactory(SslContext sslContext) throws IOException { + SSLContext context = toSSLContext(sslContext); + return context != null ? context.getServerSocketFactory() : createServerSocketFactory(); + } /** * Overriding to use SslTransportServer and allow for proper reflection. */ @Override - public TransportServer doBind(final URI location) throws IOException { - try { - Map options = new HashMap(URISupport.parseParameters(location)); - - ServerSocketFactory serverSocketFactory = createServerSocketFactory(); - SslTransportServer server = createSslTransportServer(location, (SSLServerSocketFactory)serverSocketFactory); - server.setWireFormatFactory(createWireFormatFactory(options)); - IntrospectionSupport.setProperties(server, options); - Map transportOptions = IntrospectionSupport.extractProperties(options, "transport."); - server.setTransportOption(transportOptions); - server.bind(); - - return server; - } catch (URISyntaxException e) { - throw IOExceptionSupport.create(e); - } + protected TcpTransportServer createTcpTransportServer(final URI location, ServerSocketFactory serverSocketFactory) throws IOException, URISyntaxException { + return createSslTransportServer(location, (SSLServerSocketFactory) serverSocketFactory); } /** @@ -100,26 +90,22 @@ public Transport compositeConfigure(Transport transport, WireFormat format, Map return super.compositeConfigure(transport, format, options); } + /** + * Overriding to derive the SSL socket factory from the given SslContext, + * falling back to the JVM default when none is supplied. + */ + @Override + protected SocketFactory createSocketFactory(SslContext sslContext) throws IOException { + SSLContext context = toSSLContext(sslContext); + return context != null ? context.getSocketFactory() : createSocketFactory(); + } + /** * Overriding to use SslTransports. */ @Override - protected Transport createTransport(URI location, WireFormat wf) throws UnknownHostException, IOException { - URI localLocation = null; - String path = location.getPath(); - // see if the path is a local URI location - if (path != null && path.length() > 0) { - int localPortIndex = path.indexOf(':'); - try { - Integer.parseInt(path.substring(localPortIndex + 1, path.length())); - String localString = location.getScheme() + ":/" + path; - localLocation = new URI(localString); - } catch (Exception e) { - LOG.warn("path isn't a valid local location for SslTransport to use", e); - } - } - SocketFactory socketFactory = createSocketFactory(); - return new SslTransport(wf, (SSLSocketFactory)socketFactory, location, localLocation, false); + protected TcpTransport createTcpTransport(WireFormat wf, SocketFactory socketFactory, URI location, URI localLocation) throws UnknownHostException, IOException { + return new SslTransport(wf, (SSLSocketFactory) socketFactory, location, localLocation, false); } /** @@ -131,37 +117,12 @@ protected Transport createTransport(URI location, WireFormat wf) throws UnknownH */ @Override protected ServerSocketFactory createServerSocketFactory() throws IOException { - if( SslContext.getCurrentSslContext()!=null ) { - SslContext ctx = SslContext.getCurrentSslContext(); - try { - return ctx.getSSLContext().getServerSocketFactory(); - } catch (Exception e) { - throw IOExceptionSupport.create(e); - } - } else { - return SSLServerSocketFactory.getDefault(); - } + return SSLServerSocketFactory.getDefault(); } - /** - * Creates a new SSL SocketFactory. The given factory will use user-provided - * key and trust managers (if the user provided them). - * - * @return Newly created (Ssl)SocketFactory. - * @throws IOException - */ @Override protected SocketFactory createSocketFactory() throws IOException { - if( SslContext.getCurrentSslContext()!=null ) { - SslContext ctx = SslContext.getCurrentSslContext(); - try { - return ctx.getSSLContext().getSocketFactory(); - } catch (Exception e) { - throw IOExceptionSupport.create(e); - } - } else { - return SSLSocketFactory.getDefault(); - } + return SSLSocketFactory.getDefault(); } @Override diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/tcp/TcpTransportFactory.java b/activemq-client/src/main/java/org/apache/activemq/transport/tcp/TcpTransportFactory.java index b6f42e69a62..02b83dfe04c 100644 --- a/activemq-client/src/main/java/org/apache/activemq/transport/tcp/TcpTransportFactory.java +++ b/activemq-client/src/main/java/org/apache/activemq/transport/tcp/TcpTransportFactory.java @@ -20,16 +20,17 @@ import java.net.Socket; import java.net.URI; 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; +import javax.net.ssl.SSLContext; import javax.net.ssl.SSLEngine; import org.apache.activemq.TransportLoggerSupport; +import org.apache.activemq.broker.SslContext; import org.apache.activemq.openwire.OpenWireFormat; import org.apache.activemq.transport.InactivityMonitor; import org.apache.activemq.transport.Transport; @@ -50,11 +51,25 @@ public class TcpTransportFactory extends TransportFactory { @Override public TransportServer doBind(final URI location) throws IOException { + return doBind(location, null); + } + + /** + * Binds a TCP based transport server. The given {@link SslContext} is + * handed to {@link #createServerSocketFactory(SslContext)} and + * {@link #createTcpTransportServer(URI, ServerSocketFactory, SslContext)} + * so SSL capable subclasses can derive their socket factory and server + * from it; plain TCP ignores it. The broker always binds through this + * method, so subclasses customizing the bind must override it rather + * than {@link #doBind(URI)}. + */ + @Override + public TransportServer doBind(final URI location, SslContext sslContext) throws IOException { try { Map options = new HashMap(URISupport.parseParameters(location)); - ServerSocketFactory serverSocketFactory = createServerSocketFactory(); - TcpTransportServer server = createTcpTransportServer(location, serverSocketFactory); + ServerSocketFactory serverSocketFactory = createServerSocketFactory(sslContext); + TcpTransportServer server = createTcpTransportServer(location, serverSocketFactory, sslContext); server.setWireFormatFactory(createWireFormatFactory(options)); IntrospectionSupport.setProperties(server, options); Map transportOptions = IntrospectionSupport.extractProperties(options, "transport."); @@ -81,6 +96,18 @@ protected TcpTransportServer createTcpTransportServer(final URI location, Server return new TcpTransportServer(this, location, serverSocketFactory); } + /** + * Allows SSL capable subclasses to create a TcpTransportServer that uses + * the given SslContext for accepted connections. The default ignores the + * context and delegates to + * {@link #createTcpTransportServer(URI, ServerSocketFactory)}. + * + * @param sslContext the SslContext to use, or null for the JVM default. + */ + protected TcpTransportServer createTcpTransportServer(final URI location, ServerSocketFactory serverSocketFactory, SslContext sslContext) throws IOException, URISyntaxException { + return createTcpTransportServer(location, serverSocketFactory); + } + @Override @SuppressWarnings("rawtypes") public Transport compositeConfigure(Transport transport, WireFormat format, Map options) { @@ -121,8 +148,38 @@ protected boolean isUseInactivityMonitor(Transport transport) { return true; } + /** + * Connects a TCP based transport. The given {@link SslContext} is threaded + * through {@link #doConnectInternal} to {@link #createTransport(URI, WireFormat, SslContext)} + * (and thence {@link #createSocketFactory(SslContext)}) so SSL capable + * subclasses can derive their socket factory from it; plain TCP ignores it. + * The plain {@code doConnect(URI)}/{@code doCompositeConnect(URI)} are + * inherited from {@link TransportFactory}, which routes through the same + * createTransport(URI, WireFormat, SslContext) override with a null context. + */ + @Override + public Transport doConnect(URI location, SslContext sslContext) throws IOException { + return doConnectInternal(location, sslContext, false); + } + + @Override + public Transport doCompositeConnect(URI location, SslContext sslContext) throws IOException { + return doConnectInternal(location, sslContext, true); + } + @Override - protected Transport createTransport(URI location, WireFormat wf) throws UnknownHostException, IOException { + protected Transport createTransport(URI location, WireFormat wf) throws IOException { + return createTransport(location, wf, null); + } + + /** + * Creates the client side transport for the given location, deriving the + * socket factory from the given SslContext via + * {@link #createSocketFactory(SslContext)}. + * + * @param sslContext the SslContext to use, or null for the JVM default. + */ + protected Transport createTransport(URI location, WireFormat wf, SslContext sslContext) throws IOException { URI localLocation = null; String path = location.getPath(); // see if the path is a local URI location @@ -139,7 +196,7 @@ protected Transport createTransport(URI location, WireFormat wf) throws UnknownH } } } - SocketFactory socketFactory = createSocketFactory(); + SocketFactory socketFactory = createSocketFactory(sslContext); return createTcpTransport(wf, socketFactory, location, localLocation); } @@ -163,10 +220,9 @@ public TcpTransport createTransport(WireFormat wireFormat, Socket socket, * * @return a new TcpTransport instance connected to the given location. * - * @throws UnknownHostException * @throws IOException */ - protected TcpTransport createTcpTransport(WireFormat wf, SocketFactory socketFactory, URI location, URI localLocation) throws UnknownHostException, IOException { + protected TcpTransport createTcpTransport(WireFormat wf, SocketFactory socketFactory, URI location, URI localLocation) throws IOException { return new TcpTransport(wf, socketFactory, location, localLocation); } @@ -174,10 +230,49 @@ protected ServerSocketFactory createServerSocketFactory() throws IOException { return ServerSocketFactory.getDefault(); } + /** + * Allows SSL capable subclasses to derive the ServerSocketFactory from the + * given SslContext. The default ignores the context and delegates to + * {@link #createServerSocketFactory()}. + * + * @param sslContext the SslContext to use, or null for the JVM default. + */ + protected ServerSocketFactory createServerSocketFactory(SslContext sslContext) throws IOException { + return createServerSocketFactory(); + } + + /** + * Resolves the SSLContext held by the given SslContext. + * + * @return the SSLContext, or null when no SslContext was supplied. + * @throws IOException if the SslContext cannot produce an SSLContext. + */ + protected static SSLContext toSSLContext(SslContext sslContext) throws IOException { + if (sslContext == null) { + return null; + } + try { + return sslContext.getSSLContext(); + } catch (Exception e) { + throw IOExceptionSupport.create(e); + } + } + protected SocketFactory createSocketFactory() throws IOException { return SocketFactory.getDefault(); } + /** + * Allows SSL capable subclasses to derive the SocketFactory from the given + * SslContext. The default ignores the context and delegates to + * {@link #createSocketFactory()}. + * + * @param sslContext the SslContext to use, or null for the JVM default. + */ + protected SocketFactory createSocketFactory(SslContext sslContext) throws IOException { + return createSocketFactory(); + } + protected Transport createInactivityMonitor(Transport transport, WireFormat format) { return new InactivityMonitor(transport, format); } diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/udp/UdpTransportFactory.java b/activemq-client/src/main/java/org/apache/activemq/transport/udp/UdpTransportFactory.java index 72ef4da593b..09c96ce9e07 100644 --- a/activemq-client/src/main/java/org/apache/activemq/transport/udp/UdpTransportFactory.java +++ b/activemq-client/src/main/java/org/apache/activemq/transport/udp/UdpTransportFactory.java @@ -76,7 +76,7 @@ public TransportServer doBind(final URI location) throws IOException { } @Override - public Transport configure(Transport transport, WireFormat format, Map options) throws Exception { + public Transport configure(Transport transport, WireFormat format, Map options) throws IOException { return configure(transport, format, options, false); } @@ -124,7 +124,7 @@ protected Transport createTransport(int port, WireFormat wf) throws UnknownHostE * SocketServers where new connections spin up a new separate * UDP transport */ - protected Transport configure(Transport transport, WireFormat format, Map options, boolean acceptServer) throws Exception { + protected Transport configure(Transport transport, WireFormat format, Map options, boolean acceptServer) throws IOException { IntrospectionSupport.setProperties(transport, options); UdpTransport udpTransport = (UdpTransport)transport; diff --git a/activemq-client/src/test/java/org/apache/activemq/broker/CompatibleSslContextTest.java b/activemq-client/src/test/java/org/apache/activemq/broker/CompatibleSslContextTest.java new file mode 100644 index 00000000000..8a66902afa0 --- /dev/null +++ b/activemq-client/src/test/java/org/apache/activemq/broker/CompatibleSslContextTest.java @@ -0,0 +1,199 @@ +/** + * 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.broker; + +import static org.junit.Assert.*; + +import java.security.SecureRandom; +import java.security.cert.X509Certificate; +import java.util.Arrays; +import java.util.List; + +import javax.net.ssl.KeyManager; +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManager; +import javax.net.ssl.X509TrustManager; + +import org.junit.Test; + +public class CompatibleSslContextTest { + + @Test + public void testGetSSLContextReturnsNonNull() throws Exception { + CompatibleSslContext ctx = new CompatibleSslContext(); + SSLContext sslCtx = ctx.getSSLContext(); + assertNotNull(sslCtx); + assertEquals("TLS", sslCtx.getProtocol()); + } + + @Test + public void testGetSSLContextIsSingleton() throws Exception { + CompatibleSslContext ctx = new CompatibleSslContext(); + SSLContext first = ctx.getSSLContext(); + SSLContext second = ctx.getSSLContext(); + assertSame(first, second); + } + + @Test + public void testArrayConstructor() throws Exception { + TrustManager tm = new PermissiveTrustManager(); + CompatibleSslContext ctx = new CompatibleSslContext(null, new TrustManager[]{tm}, null); + + List list = ctx.getTrustManagers(); + assertEquals(1, list.size()); + assertSame(tm, list.get(0)); + assertTrue(ctx.getKeyManagers().isEmpty()); + } + + @Test + public void testListBasedGettersAndSetters() { + CompatibleSslContext ctx = new CompatibleSslContext(); + + assertTrue(ctx.getKeyManagers().isEmpty()); + assertTrue(ctx.getTrustManagers().isEmpty()); + + TrustManager tm = new PermissiveTrustManager(); + ctx.setTrustManagers(Arrays.asList(tm)); + assertEquals(1, ctx.getTrustManagers().size()); + + TrustManager[] array = ctx.getTrustManagersAsArray(); + assertEquals(1, array.length); + assertSame(tm, array[0]); + } + + @Test + public void testAddRemoveKeyManager() { + CompatibleSslContext ctx = new CompatibleSslContext(); + KeyManager km = new DummyKeyManager(); + + ctx.addKeyManager(km); + assertEquals(1, ctx.getKeyManagers().size()); + assertEquals(1, ctx.getKeyManagersAsArray().length); + + assertTrue(ctx.removeKeyManager(km)); + assertTrue(ctx.getKeyManagers().isEmpty()); + } + + @Test + public void testAddRemoveTrustManager() { + CompatibleSslContext ctx = new CompatibleSslContext(); + TrustManager tm = new PermissiveTrustManager(); + + ctx.addTrustManager(tm); + assertEquals(1, ctx.getTrustManagers().size()); + assertEquals(1, ctx.getTrustManagersAsArray().length); + + assertTrue(ctx.removeTrustManager(tm)); + assertTrue(ctx.getTrustManagers().isEmpty()); + } + + @Test + public void testProtocolOverride() throws Exception { + CompatibleSslContext ctx = new CompatibleSslContext(); + ctx.setProtocol("TLSv1.2"); + assertEquals("TLSv1.2", ctx.getProtocol()); + + SSLContext sslCtx = ctx.getSSLContext(); + assertEquals("TLSv1.2", sslCtx.getProtocol()); + } + + @Test + public void testSetSSLContextDirectly() throws Exception { + CompatibleSslContext ctx = new CompatibleSslContext(); + SSLContext manual = SSLContext.getInstance("TLS"); + manual.init(null, null, null); + ctx.setSSLContext(manual); + + assertSame(manual, ctx.getSSLContext()); + } + + @Test + public void testBeanProperties() { + CompatibleSslContext ctx = new CompatibleSslContext(); + + assertEquals("TLS", ctx.getProtocol()); + assertNull(ctx.getProvider()); + assertNull(ctx.getSecureRandom()); + + ctx.setProvider("SunJSSE"); + assertEquals("SunJSSE", ctx.getProvider()); + + SecureRandom sr = new SecureRandom(); + ctx.setSecureRandom(sr); + assertSame(sr, ctx.getSecureRandom()); + } + + @Test + public void testNoThreadLocalSideEffects() throws Exception { + SslContext before = SslContext.getCurrentSslContext(); + CompatibleSslContext ctx = new CompatibleSslContext(); + ctx.getSSLContext(); + assertSame("CompatibleSslContext must not touch ThreadLocal", + before, SslContext.getCurrentSslContext()); + } + + @Test + public void testProtectedFieldAccessFromSubclass() throws Exception { + CompatibleSslContext ctx = new CompatibleSslContext(); + + ctx.keyManagers.add(new DummyKeyManager()); + ctx.trustManagers.add(new PermissiveTrustManager()); + ctx.secureRandom = new SecureRandom(); + + assertEquals(1, ctx.getKeyManagersAsArray().length); + assertEquals(1, ctx.getTrustManagersAsArray().length); + assertNotNull(ctx.getSecureRandom()); + } + + @Test + public void testReloadIsNoOp() throws Exception { + CompatibleSslContext ctx = new CompatibleSslContext(); + SSLContext before = ctx.getSSLContext(); + ctx.reload(); + assertSame(before, ctx.getSSLContext()); + } + + @Test + public void testApiCompatibilityWithThreadLocalSslContext() throws Exception { + ThreadLocalSslContext threadLocal = new ThreadLocalSslContext(); + CompatibleSslContext compatible = new CompatibleSslContext(); + + threadLocal.setProtocol("TLSv1.2"); + compatible.setProtocol("TLSv1.2"); + + TrustManager tm = new PermissiveTrustManager(); + threadLocal.addTrustManager(tm); + compatible.addTrustManager(tm); + + assertEquals(threadLocal.getTrustManagers().size(), compatible.getTrustManagers().size()); + assertEquals(threadLocal.getTrustManagersAsArray().length, compatible.getTrustManagersAsArray().length); + assertEquals(threadLocal.getProtocol(), compatible.getProtocol()); + assertEquals(threadLocal.getSSLContext().getProtocol(), compatible.getSSLContext().getProtocol()); + } + + private static class PermissiveTrustManager implements X509TrustManager { + @Override + public void checkClientTrusted(X509Certificate[] chain, String authType) {} + @Override + public void checkServerTrusted(X509Certificate[] chain, String authType) {} + @Override + public X509Certificate[] getAcceptedIssuers() { return new X509Certificate[0]; } + } + + private static class DummyKeyManager implements KeyManager { + } +} diff --git a/activemq-client/src/test/java/org/apache/activemq/broker/DefaultSslContextTest.java b/activemq-client/src/test/java/org/apache/activemq/broker/DefaultSslContextTest.java new file mode 100644 index 00000000000..59df7f4fa31 --- /dev/null +++ b/activemq-client/src/test/java/org/apache/activemq/broker/DefaultSslContextTest.java @@ -0,0 +1,120 @@ +/** + * 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.broker; + +import static org.junit.Assert.*; + +import java.security.SecureRandom; + +import javax.net.ssl.KeyManager; +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManager; +import javax.net.ssl.X509TrustManager; +import java.security.cert.X509Certificate; + +import org.junit.Test; + +public class DefaultSslContextTest { + + @Test + public void testGetSSLContextReturnsNonNull() throws Exception { + DefaultSslContext ctx = new DefaultSslContext(); + SSLContext sslCtx = ctx.getSSLContext(); + assertNotNull(sslCtx); + assertEquals("TLS", sslCtx.getProtocol()); + } + + @Test + public void testGetSSLContextIsSingleton() throws Exception { + DefaultSslContext ctx = new DefaultSslContext(); + SSLContext first = ctx.getSSLContext(); + SSLContext second = ctx.getSSLContext(); + assertSame(first, second); + } + + @Test + public void testConstructorWithManagers() throws Exception { + TrustManager tm = new PermissiveTrustManager(); + DefaultSslContext ctx = new DefaultSslContext(null, new TrustManager[]{tm}, null); + + SSLContext sslCtx = ctx.getSSLContext(); + assertNotNull(sslCtx); + assertArrayEquals(new TrustManager[]{tm}, ctx.getTrustManagers()); + } + + @Test + public void testProtocolOverride() throws Exception { + DefaultSslContext ctx = new DefaultSslContext(); + ctx.setProtocol("TLSv1.2"); + assertEquals("TLSv1.2", ctx.getProtocol()); + + SSLContext sslCtx = ctx.getSSLContext(); + assertEquals("TLSv1.2", sslCtx.getProtocol()); + } + + @Test + public void testReloadIsNoOp() throws Exception { + DefaultSslContext ctx = new DefaultSslContext(); + SSLContext before = ctx.getSSLContext(); + ctx.reload(); + assertSame(before, ctx.getSSLContext()); + } + + @Test + public void testBeanProperties() { + DefaultSslContext ctx = new DefaultSslContext(); + + assertNull(ctx.getKeyManagers()); + assertNull(ctx.getTrustManagers()); + assertNull(ctx.getSecureRandom()); + assertNull(ctx.getProvider()); + assertEquals("TLS", ctx.getProtocol()); + + KeyManager[] kms = new KeyManager[0]; + ctx.setKeyManagers(kms); + assertSame(kms, ctx.getKeyManagers()); + + TrustManager[] tms = new TrustManager[0]; + ctx.setTrustManagers(tms); + assertSame(tms, ctx.getTrustManagers()); + + SecureRandom sr = new SecureRandom(); + ctx.setSecureRandom(sr); + assertSame(sr, ctx.getSecureRandom()); + + ctx.setProvider("SunJSSE"); + assertEquals("SunJSSE", ctx.getProvider()); + } + + @Test + public void testNoThreadLocalSideEffects() throws Exception { + SslContext before = SslContext.getCurrentSslContext(); + DefaultSslContext ctx = new DefaultSslContext(); + ctx.getSSLContext(); + assertSame("DefaultSslContext must not touch ThreadLocal", + before, SslContext.getCurrentSslContext()); + } + + private static class PermissiveTrustManager implements X509TrustManager { + @Override + public void checkClientTrusted(X509Certificate[] chain, String authType) {} + @Override + public void checkServerTrusted(X509Certificate[] chain, String authType) {} + @Override + public X509Certificate[] getAcceptedIssuers() { return new X509Certificate[0]; } + } +} diff --git a/activemq-client/src/test/java/org/apache/activemq/broker/scheduler/SslContextTest.java b/activemq-client/src/test/java/org/apache/activemq/broker/scheduler/SslContextTest.java index f0405b04500..da4f531d84b 100644 --- a/activemq-client/src/test/java/org/apache/activemq/broker/scheduler/SslContextTest.java +++ b/activemq-client/src/test/java/org/apache/activemq/broker/scheduler/SslContextTest.java @@ -17,6 +17,7 @@ package org.apache.activemq.broker.scheduler; +import org.apache.activemq.broker.DefaultSslContext; import org.apache.activemq.broker.SslContext; import org.junit.Test; @@ -32,7 +33,7 @@ import static org.junit.Assert.assertTrue; public class SslContextTest { - SslContext underTest = new SslContext(); + SslContext underTest = new DefaultSslContext(); @Test public void testConcurrentGet() throws Exception { diff --git a/activemq-http/src/main/java/org/apache/activemq/transport/http/HttpTransportFactory.java b/activemq-http/src/main/java/org/apache/activemq/transport/http/HttpTransportFactory.java index 4b53c3151c9..e6425fa9d2f 100644 --- a/activemq-http/src/main/java/org/apache/activemq/transport/http/HttpTransportFactory.java +++ b/activemq-http/src/main/java/org/apache/activemq/transport/http/HttpTransportFactory.java @@ -23,6 +23,7 @@ import java.util.HashMap; import java.util.Map; +import org.apache.activemq.broker.SslContext; import org.apache.activemq.transport.Transport; import org.apache.activemq.transport.TransportFactory; import org.apache.activemq.transport.TransportLoggerFactory; @@ -87,9 +88,28 @@ protected Transport createTransport(URI location, WireFormat wf) throws IOExcept return new HttpClientTransport(textWireFormat, uri); } + /** + * HttpTransportFactory extends TransportFactory directly (not + * TcpTransportFactory), so it does not inherit Tcp's SslContext-threading + * doConnect. It must supply its own, or the base doConnect(URI, SslContext) + * would silently drop the context and an HTTPS client would fall back to the + * JVM default trust store. The context is threaded through doConnectInternal + * to createTransport(URI, WireFormat, SslContext) — plain HTTP ignores it, + * HttpsTransportFactory overrides that method to use it. + */ + @Override + public Transport doConnect(URI location, SslContext sslContext) throws IOException { + return doConnectInternal(location, sslContext, false); + } + + @Override + public Transport doCompositeConnect(URI location, SslContext sslContext) throws IOException { + return doConnectInternal(location, sslContext, true); + } + @Override @SuppressWarnings("rawtypes") - public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) throws Exception { + public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) { return compositeConfigure(transport, format, options); } diff --git a/activemq-http/src/main/java/org/apache/activemq/transport/https/HttpsClientTransport.java b/activemq-http/src/main/java/org/apache/activemq/transport/https/HttpsClientTransport.java index d69f227bdd1..fbb0e2b6f3e 100644 --- a/activemq-http/src/main/java/org/apache/activemq/transport/https/HttpsClientTransport.java +++ b/activemq-http/src/main/java/org/apache/activemq/transport/https/HttpsClientTransport.java @@ -42,10 +42,18 @@ public class HttpsClientTransport extends HttpClientTransport { private boolean verifyHostName = true; public HttpsClientTransport(TextWireFormat wireFormat, URI remoteUrl) { + this(wireFormat, remoteUrl, null); + } + + public HttpsClientTransport(TextWireFormat wireFormat, URI remoteUrl, SslContext sslContext) { super(wireFormat, remoteUrl); try { - sslSocketFactory = createSocketFactory(); - } catch (IOException e) { + if (sslContext != null) { + sslSocketFactory = sslContext.getSSLContext().getSocketFactory(); + } else { + sslSocketFactory = (javax.net.ssl.SSLSocketFactory) javax.net.ssl.SSLSocketFactory.getDefault(); + } + } catch (Exception e) { throw new IllegalStateException("Error trying to configure TLS", e); } } @@ -69,25 +77,9 @@ private Registry createRegistry() { } } - /** - * Creates a new SSL SocketFactory. The given factory will use user-provided - * key and trust managers (if the user provided them). - * - * @return Newly created (Ssl)SocketFactory. - * @throws IOException - */ + @Deprecated protected javax.net.ssl.SSLSocketFactory createSocketFactory() throws IOException { - if (SslContext.getCurrentSslContext() != null) { - SslContext ctx = SslContext.getCurrentSslContext(); - try { - return ctx.getSSLContext().getSocketFactory(); - } catch (Exception e) { - throw IOExceptionSupport.create(e); - } - } else { - return (javax.net.ssl.SSLSocketFactory) javax.net.ssl.SSLSocketFactory.getDefault(); - } - + return (javax.net.ssl.SSLSocketFactory) javax.net.ssl.SSLSocketFactory.getDefault(); } @Override diff --git a/activemq-http/src/main/java/org/apache/activemq/transport/https/HttpsTransportFactory.java b/activemq-http/src/main/java/org/apache/activemq/transport/https/HttpsTransportFactory.java index d8f5ea2cd2f..cfd06ee408c 100644 --- a/activemq-http/src/main/java/org/apache/activemq/transport/https/HttpsTransportFactory.java +++ b/activemq-http/src/main/java/org/apache/activemq/transport/https/HttpsTransportFactory.java @@ -43,9 +43,14 @@ public TransportServer doBind(String brokerId, URI location) throws IOException @Override public TransportServer doBind(URI location) throws IOException { + return doBind(location, null); + } + + @Override + public TransportServer doBind(URI location, SslContext sslContext) throws IOException { try { Map options = new HashMap(URISupport.parseParameters(location)); - HttpsTransportServer result = new HttpsTransportServer(location, this, SslContext.getCurrentSslContext()); + HttpsTransportServer result = new HttpsTransportServer(location, this, sslContext); Map httpOptions = IntrospectionSupport.extractProperties(options, "http."); Map transportOptions = IntrospectionSupport.extractProperties(options, "transport."); result.setTransportOption(transportOptions); @@ -58,7 +63,18 @@ public TransportServer doBind(URI location) throws IOException { @Override protected Transport createTransport(URI location, WireFormat wf) throws IOException { - // need to remove options from uri + return createTransport(location, wf, null); + } + + /** + * HTTPS SSL-aware createTransport override: builds the client transport with the supplied SslContext + * (null = JVM default). The connect template threads the context here via + * TcpTransportFactory's inherited doConnect(URI, SslContext); overriding this + * 3-arg (instead of the 2-arg) is what lets HTTPS receive the context + * directly, replacing the old sslContext field + doConnect stash. + */ + @Override + protected Transport createTransport(URI location, WireFormat wf, SslContext sslContext) throws IOException { try { URI uri = URISupport.removeQuery(location); @@ -69,7 +85,7 @@ protected Transport createTransport(URI location, WireFormat wf) throws IOExcept verifyHostName = Boolean.parseBoolean(transportOptions.get("verifyHostName").toString()); } - HttpsClientTransport clientTransport = new HttpsClientTransport(asTextWireFormat(wf), uri); + HttpsClientTransport clientTransport = new HttpsClientTransport(asTextWireFormat(wf), uri, sslContext); clientTransport.setVerifyHostName(verifyHostName); return clientTransport; } catch (URISyntaxException e) { diff --git a/activemq-http/src/main/java/org/apache/activemq/transport/wss/WSSTransportFactory.java b/activemq-http/src/main/java/org/apache/activemq/transport/wss/WSSTransportFactory.java index 05a8159ca52..44b00e2b8e7 100644 --- a/activemq-http/src/main/java/org/apache/activemq/transport/wss/WSSTransportFactory.java +++ b/activemq-http/src/main/java/org/apache/activemq/transport/wss/WSSTransportFactory.java @@ -40,9 +40,14 @@ public class WSSTransportFactory extends TransportFactory implements BrokerServi @Override public TransportServer doBind(URI location) throws IOException { + return doBind(location, null); + } + + @Override + public TransportServer doBind(URI location, SslContext sslContext) throws IOException { try { Map options = new HashMap(URISupport.parseParameters(location)); - WSSTransportServer result = new WSSTransportServer(location, SslContext.getCurrentSslContext()); + WSSTransportServer result = new WSSTransportServer(location, sslContext); Map httpOptions = IntrospectionSupport.extractProperties(options, "http."); Map transportOptions = IntrospectionSupport.extractProperties(options, ""); IntrospectionSupport.setProperties(result, transportOptions); diff --git a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTNIOSSLTransportFactory.java b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTNIOSSLTransportFactory.java index 87fc48eada7..1b4617df7d3 100644 --- a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTNIOSSLTransportFactory.java +++ b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTNIOSSLTransportFactory.java @@ -30,20 +30,22 @@ import org.apache.activemq.broker.SslContext; import org.apache.activemq.transport.Transport; -import org.apache.activemq.transport.TransportServer; import org.apache.activemq.transport.nio.NIOSSLTransportServer; 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.IntrospectionSupport; import org.apache.activemq.wireformat.WireFormat; public class MQTTNIOSSLTransportFactory extends MQTTNIOTransportFactory { - SSLContext context; - @Override protected TcpTransportServer createTcpTransportServer(URI location, ServerSocketFactory serverSocketFactory) throws IOException, URISyntaxException { + return createTcpTransportServer(location, serverSocketFactory, null); + } + + @Override + protected TcpTransportServer createTcpTransportServer(URI location, ServerSocketFactory serverSocketFactory, SslContext sslContext) throws IOException, URISyntaxException { + final SSLContext context = toSSLContext(sslContext); NIOSSLTransportServer result = new NIOSSLTransportServer(context, this, location, serverSocketFactory) { @Override protected Transport createTransport(Socket socket, WireFormat format) throws IOException { @@ -73,17 +75,4 @@ public TcpTransport createTransport(WireFormat wireFormat, Socket socket, throws IOException { return new MQTTNIOSSLTransport(wireFormat, socket, engine, initBuffer, inputBuffer); } - - @Override - public TransportServer doBind(URI location) throws IOException { - if (SslContext.getCurrentSslContext() != null) { - try { - context = SslContext.getCurrentSslContext().getSSLContext(); - } catch (Exception e) { - throw new IOException(e); - } - } - return super.doBind(location); - } - } diff --git a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTNIOTransportFactory.java b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTNIOTransportFactory.java index 5823add2045..33872aef275 100644 --- a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTNIOTransportFactory.java +++ b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTNIOTransportFactory.java @@ -75,7 +75,7 @@ public TcpTransport createTransport(WireFormat wireFormat, Socket socket, @SuppressWarnings("rawtypes") @Override - public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) throws Exception { + public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) { transport = super.serverConfigure(transport, format, options); MutexTransport mutex = transport.narrow(MutexTransport.class); diff --git a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTSslTransportFactory.java b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTSslTransportFactory.java index 99353879190..feb74c816cb 100644 --- a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTSslTransportFactory.java +++ b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTSslTransportFactory.java @@ -61,7 +61,7 @@ protected SslTransportServer createSslTransportServer(URI location, SSLServerSoc @SuppressWarnings("rawtypes") @Override - public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) throws Exception { + public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) { transport = super.serverConfigure(transport, format, options); MutexTransport mutex = transport.narrow(MutexTransport.class); diff --git a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTTransportFactory.java b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTTransportFactory.java index 47afedb2499..91c15345cc9 100644 --- a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTTransportFactory.java +++ b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTTransportFactory.java @@ -67,7 +67,7 @@ public void setBrokerService(BrokerService brokerService) { @SuppressWarnings("rawtypes") @Override - public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) throws Exception { + public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) { transport = super.serverConfigure(transport, format, options); MutexTransport mutex = transport.narrow(MutexTransport.class); diff --git a/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/util/ResourceLoadingSslContext.java b/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/util/ResourceLoadingSslContext.java index b1b8375aee4..a57e61e8d0a 100644 --- a/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/util/ResourceLoadingSslContext.java +++ b/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/util/ResourceLoadingSslContext.java @@ -32,7 +32,7 @@ import javax.net.ssl.TrustManager; import javax.net.ssl.TrustManagerFactory; -import org.apache.activemq.broker.SslContext; +import org.apache.activemq.broker.CompatibleSslContext; import org.springframework.core.io.ClassPathResource; import org.springframework.core.io.FileSystemResource; import org.springframework.core.io.Resource; @@ -42,7 +42,7 @@ /** * Extends the SslContext so that it's easier to configure from spring. */ -public class ResourceLoadingSslContext extends SslContext { +public class ResourceLoadingSslContext extends CompatibleSslContext { private String keyStoreType = "jks"; private String trustStoreType = "jks"; diff --git a/activemq-ra/src/test/java/org/apache/activemq/ra/SSLMAnagedConnectionFactoryTest.java b/activemq-ra/src/test/java/org/apache/activemq/ra/SSLMAnagedConnectionFactoryTest.java index 83a94a63cb1..b4a6f373c17 100644 --- a/activemq-ra/src/test/java/org/apache/activemq/ra/SSLMAnagedConnectionFactoryTest.java +++ b/activemq-ra/src/test/java/org/apache/activemq/ra/SSLMAnagedConnectionFactoryTest.java @@ -24,10 +24,7 @@ import javax.net.ssl.TrustManager; import org.apache.activemq.broker.SslBrokerService; -import org.apache.activemq.broker.SslContext; import org.apache.activemq.broker.TransportConnector; -import org.apache.activemq.transport.TransportFactory; -import org.apache.activemq.transport.tcp.SslTransportFactory; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -93,9 +90,8 @@ private void createAndStartBroker() throws Exception { connectionURI = connector.getPublishableConnectString(); - SslTransportFactory sslFactory = new SslTransportFactory(); - SslContext ctx = new SslContext(km, tm, null); - SslContext.setCurrentSslContext(ctx); - TransportFactory.registerTransportFactory("ssl", sslFactory); + // No ThreadLocal SslContext needed: the RA client uses + // ActiveMQSslConnectionFactory configured with the trust/key stores + // set on the managed connection factory above. } } diff --git a/activemq-ra/src/test/java/org/apache/activemq/ra/SSLTest.java b/activemq-ra/src/test/java/org/apache/activemq/ra/SSLTest.java index 2aa172eb388..37f3e2c2920 100644 --- a/activemq-ra/src/test/java/org/apache/activemq/ra/SSLTest.java +++ b/activemq-ra/src/test/java/org/apache/activemq/ra/SSLTest.java @@ -64,13 +64,10 @@ import org.apache.activemq.ActiveMQSslConnectionFactory; import org.apache.activemq.advisory.AdvisorySupport; import org.apache.activemq.broker.SslBrokerService; -import org.apache.activemq.broker.SslContext; import org.apache.activemq.broker.TransportConnector; import org.apache.activemq.command.ActiveMQMessage; import org.apache.activemq.command.ActiveMQQueue; import org.apache.activemq.command.ConsumerInfo; -import org.apache.activemq.transport.TransportFactory; -import org.apache.activemq.transport.tcp.SslTransportFactory; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -112,12 +109,11 @@ private void createAndStartBroker() throws Exception { TrustManager[] tm = getTrustManager(); connector = broker.addSslConnector(BIND_ADDRESS, km, tm, null); broker.start(); - broker.waitUntilStarted(); // for client side + broker.waitUntilStarted(); - SslTransportFactory sslFactory = new SslTransportFactory(); - SslContext ctx = new SslContext(km, tm, null); - SslContext.setCurrentSslContext(ctx); - TransportFactory.registerTransportFactory("ssl", sslFactory); + // No ThreadLocal SslContext needed: every client in this test + // (ActiveMQSslConnectionFactory, the RA, and the activation specs) + // configures its trust/key stores explicitly. } private static final class StubBootstrapContext implements BootstrapContext { diff --git a/activemq-spring/src/main/java/org/apache/activemq/spring/SpringSslContext.java b/activemq-spring/src/main/java/org/apache/activemq/spring/SpringSslContext.java index adb7f58a820..3a25feb0053 100644 --- a/activemq-spring/src/main/java/org/apache/activemq/spring/SpringSslContext.java +++ b/activemq-spring/src/main/java/org/apache/activemq/spring/SpringSslContext.java @@ -30,7 +30,7 @@ import jakarta.annotation.PostConstruct; import javax.net.ssl.*; -import org.apache.activemq.broker.SslContext; +import org.apache.activemq.broker.CompatibleSslContext; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.core.io.Resource; @@ -42,7 +42,7 @@ * * */ -public class SpringSslContext extends SslContext { +public class SpringSslContext extends CompatibleSslContext { private static final transient Logger LOG = LoggerFactory.getLogger(SpringSslContext.class); diff --git a/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompNIOSSLTransportFactory.java b/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompNIOSSLTransportFactory.java index 27a871224d2..252d4efa7bc 100644 --- a/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompNIOSSLTransportFactory.java +++ b/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompNIOSSLTransportFactory.java @@ -30,7 +30,6 @@ import org.apache.activemq.broker.SslContext; import org.apache.activemq.transport.Transport; -import org.apache.activemq.transport.TransportServer; import org.apache.activemq.transport.nio.NIOSSLTransportServer; import org.apache.activemq.transport.tcp.TcpTransport; import org.apache.activemq.transport.tcp.TcpTransport.InitBuffer; @@ -39,10 +38,14 @@ public class StompNIOSSLTransportFactory extends StompNIOTransportFactory { - protected SSLContext context; - @Override protected TcpTransportServer createTcpTransportServer(URI location, ServerSocketFactory serverSocketFactory) throws IOException, URISyntaxException { + return createTcpTransportServer(location, serverSocketFactory, null); + } + + @Override + protected TcpTransportServer createTcpTransportServer(URI location, ServerSocketFactory serverSocketFactory, SslContext sslContext) throws IOException, URISyntaxException { + final SSLContext context = toSSLContext(sslContext); return new NIOSSLTransportServer(context, this, location, serverSocketFactory) { @Override @@ -71,16 +74,4 @@ public TcpTransport createTransport(WireFormat wireFormat, Socket socket, throws IOException { return new StompNIOSSLTransport(wireFormat, socket, engine, initBuffer, inputBuffer); } - - @Override - public TransportServer doBind(URI location) throws IOException { - if (SslContext.getCurrentSslContext() != null) { - try { - context = SslContext.getCurrentSslContext().getSSLContext(); - } catch (Exception e) { - throw new IOException(e); - } - } - return super.doBind(location); - } } diff --git a/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompNIOTransportFactory.java b/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompNIOTransportFactory.java index be6768decc6..7de94235030 100644 --- a/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompNIOTransportFactory.java +++ b/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompNIOTransportFactory.java @@ -76,7 +76,7 @@ public TcpTransport createTransport(WireFormat wireFormat, Socket socket, @SuppressWarnings("rawtypes") @Override - public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) throws Exception { + public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) { transport = super.serverConfigure(transport, format, options); MutexTransport mutex = transport.narrow(MutexTransport.class); diff --git a/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompSslTransportFactory.java b/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompSslTransportFactory.java index 7fd1b4b92d2..1be899de3dd 100644 --- a/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompSslTransportFactory.java +++ b/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompSslTransportFactory.java @@ -107,7 +107,7 @@ public Transport compositeConfigure(Transport transport, WireFormat format, Map @SuppressWarnings("rawtypes") @Override - public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) throws Exception { + public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) { transport = super.serverConfigure(transport, format, options); MutexTransport mutex = transport.narrow(MutexTransport.class); diff --git a/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompTransportFactory.java b/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompTransportFactory.java index 4fef5d3e05c..fa7b5c96601 100644 --- a/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompTransportFactory.java +++ b/activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompTransportFactory.java @@ -52,7 +52,7 @@ public void setBrokerService(BrokerService brokerService) { @SuppressWarnings("rawtypes") @Override - public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) throws Exception { + public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) { transport = super.serverConfigure(transport, format, options); MutexTransport mutex = transport.narrow(MutexTransport.class); diff --git a/activemq-stomp/src/test/java/org/apache/activemq/transport/stomp/util/ResourceLoadingSslContext.java b/activemq-stomp/src/test/java/org/apache/activemq/transport/stomp/util/ResourceLoadingSslContext.java index 3bbfefc07bb..50a3a0c8458 100644 --- a/activemq-stomp/src/test/java/org/apache/activemq/transport/stomp/util/ResourceLoadingSslContext.java +++ b/activemq-stomp/src/test/java/org/apache/activemq/transport/stomp/util/ResourceLoadingSslContext.java @@ -32,7 +32,7 @@ import javax.net.ssl.TrustManager; import javax.net.ssl.TrustManagerFactory; -import org.apache.activemq.broker.SslContext; +import org.apache.activemq.broker.CompatibleSslContext; import org.springframework.core.io.ClassPathResource; import org.springframework.core.io.FileSystemResource; import org.springframework.core.io.Resource; @@ -42,7 +42,7 @@ /** * Extends the SslContext so that it's easier to configure from spring. */ -public class ResourceLoadingSslContext extends SslContext { +public class ResourceLoadingSslContext extends CompatibleSslContext { private String keyStoreType="jks"; private String trustStoreType="jks"; diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/network/FailoverStaticNetworkTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/network/FailoverStaticNetworkTest.java index 03665155093..aebc04b2896 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/network/FailoverStaticNetworkTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/network/FailoverStaticNetworkTest.java @@ -40,10 +40,11 @@ import javax.net.ssl.KeyManager; import javax.net.ssl.TrustManager; -import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.ActiveMQSslConnectionFactory; import org.apache.activemq.AutoFailTestSupport; import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.SslContext; +import org.apache.activemq.broker.DefaultSslContext; import org.apache.activemq.broker.TransportConnector; import org.apache.activemq.command.ActiveMQDestination; import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter; @@ -68,6 +69,8 @@ public class FailoverStaticNetworkTest { private SslContext sslContext; + private KeyManager[] km; + private TrustManager[] tm; protected BrokerService createBroker(String scheme, String listenPort, String[] networkToPorts) throws Exception { return createBroker(scheme, listenPort, networkToPorts, null); @@ -118,9 +121,9 @@ private BrokerService createBroker(String listenPort, String dataDir) throws Exc @Before public void setUp() throws Exception { - KeyManager[] km = SslBrokerServiceTest.getKeyManager(); - TrustManager[] tm = SslBrokerServiceTest.getTrustManager(); - sslContext = new SslContext(km, tm, null); + km = SslBrokerServiceTest.getKeyManager(); + tm = SslBrokerServiceTest.getTrustManager(); + sslContext = new DefaultSslContext(km, tm, null); } @After @@ -418,7 +421,6 @@ private void doTestNetworkSendReceive(final BrokerService to, final BrokerServic LOG.info("Creating Consumer on the networked broker ..." + from); - SslContext.setCurrentSslContext(sslContext); // Create a consumer on brokerA ConnectionFactory consFactory = createConnectionFactory(from); Connection consConn = consFactory.createConnection(); @@ -460,7 +462,10 @@ private void sendMessageTo(ActiveMQDestination destination, BrokerService broker protected ConnectionFactory createConnectionFactory(final BrokerService broker) throws Exception { String url = broker.getTransportConnectors().get(0).getServer().getConnectURI().toString(); - ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url); + // explicit key/trust managers replace the previous ThreadLocal + // SslContext propagation; ignored by the tcp scheme variants + ActiveMQSslConnectionFactory connectionFactory = new ActiveMQSslConnectionFactory(url); + connectionFactory.setKeyAndTrustManagers(km, tm, null); connectionFactory.setOptimizedMessageDispatch(true); connectionFactory.setDispatchAsync(false); connectionFactory.setUseAsyncSend(false); diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkReconnectSslNioTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkReconnectSslNioTest.java index 7feeb892491..cbe4ab04545 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkReconnectSslNioTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkReconnectSslNioTest.java @@ -18,6 +18,7 @@ import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.SslContext; +import org.apache.activemq.broker.DefaultSslContext; import org.apache.activemq.broker.TransportConnection; import org.apache.activemq.broker.TransportConnector; import org.apache.activemq.command.ConnectionError; @@ -44,7 +45,7 @@ public class NetworkReconnectSslNioTest { @Test public void testForceReconnect() throws Exception { - final SslContext sslContext = new SslContext(getKeyManager(), getTrustManager(), null); + final SslContext sslContext = new DefaultSslContext(getKeyManager(), getTrustManager(), null); BrokerService remote = new BrokerService(); remote.setBrokerName("R"); diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/transport/tcp/SslBrokerServiceTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/transport/tcp/SslBrokerServiceTest.java index 57406689c34..8dc92172e57 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/transport/tcp/SslBrokerServiceTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/transport/tcp/SslBrokerServiceTest.java @@ -38,10 +38,8 @@ import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.SslBrokerService; -import org.apache.activemq.broker.SslContext; import org.apache.activemq.broker.TransportConnector; import org.apache.activemq.transport.TransportBrokerTestSupport; -import org.apache.activemq.transport.TransportFactory; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.junit.experimental.categories.Category; @@ -74,25 +72,35 @@ protected BrokerService createBroker() throws Exception { limitedCipherSuites = service.addSslConnector("ssl://localhost:0?transport.enabledCipherSuites=TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256", km, tm, null); needClientAuthConnector = service.addSslConnector("ssl://localhost:0?transport.needClientAuth=true", km, tm, null); - // for client side - SslTransportFactory sslFactory = new SslTransportFactory(); - SslContext ctx = new SslContext(km, tm, null); - SslContext.setCurrentSslContext(ctx); - TransportFactory.registerTransportFactory("ssl", sslFactory); - + // for client side: the inherited createConnection() path uses the + // JVM default SSLSocketFactory, configured via system properties + // (same pattern as SslTransportBrokerTest) + System.setProperty("javax.net.ssl.trustStore", SslTransportBrokerTest.TRUST_KEYSTORE); + System.setProperty("javax.net.ssl.trustStorePassword", SslTransportBrokerTest.PASSWORD); + System.setProperty("javax.net.ssl.trustStoreType", SslTransportBrokerTest.KEYSTORE_TYPE); + System.setProperty("javax.net.ssl.keyStore", SslTransportBrokerTest.SERVER_KEYSTORE); + System.setProperty("javax.net.ssl.keyStorePassword", SslTransportBrokerTest.PASSWORD); + System.setProperty("javax.net.ssl.keyStoreType", SslTransportBrokerTest.KEYSTORE_TYPE); + return service; } public void testNeedClientAuthReject() throws Exception { - SSLContext context = SSLContext.getInstance("TLS"); + // TLSv1.2: under TLS 1.3 the client finishes its handshake before the + // server's certificate_required alert arrives, so the rejection would + // only surface on first read/write rather than in startHandshake() + SSLContext context = SSLContext.getInstance("TLSv1.2"); // no client cert context.init(null, getTrustManager(), null); try { makeSSLConnection(context, null, needClientAuthConnector); fail("expected failure on no client cert"); - } catch (SSLException expected) { - expected.printStackTrace(); + } catch (SSLException | SocketException expected) { + // SSLException when the server's fatal alert is read first; + // SocketException (broken pipe/reset) when the client's handshake + // write races the server-side close after the alert + LOG.info("Got expected rejection: {}", expected.toString()); } // should work with regular connector makeSSLConnection(context, null, connector); @@ -127,12 +135,14 @@ private void makeSSLConnection(SSLContext context, String enabledSuites[], Trans sslSocket.setEnabledCipherSuites(enabledSuites); } sslSocket.setSoTimeout(5000); - - SSLSession session = sslSocket.getSession(); + + // handshake first so failures surface as SSLException (getSession() + // swallows handshake errors and returns an invalid session) sslSocket.startHandshake(); + SSLSession session = sslSocket.getSession(); LOG.info("cyphersuite: " + session.getCipherSuite()); LOG.info("peer port: " + session.getPeerPort()); - LOG.info("peer cert: " + session.getPeerCertificateChain()[0].toString()); + LOG.info("peer cert: " + session.getPeerCertificates()[0].toString()); } public static TrustManager[] getTrustManager() throws Exception { diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/transport/tcp/SslContextBrokerServiceTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/transport/tcp/SslContextBrokerServiceTest.java index a966f23cdeb..c6c977be227 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/transport/tcp/SslContextBrokerServiceTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/transport/tcp/SslContextBrokerServiceTest.java @@ -23,6 +23,7 @@ import junit.framework.TestCase; import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.broker.CompatibleSslContext; import org.apache.activemq.broker.TransportConnector; import org.springframework.context.support.ClassPathXmlApplicationContext; @@ -43,8 +44,9 @@ public void testConfiguration() throws URISyntaxException { assertEquals(new URI("ssl://localhost:61616"), connector.getUri()); assertNotNull(broker.getSslContext()); - assertFalse(broker.getSslContext().getKeyManagers().isEmpty()); - assertFalse(broker.getSslContext().getTrustManagers().isEmpty()); + CompatibleSslContext sslCtx = (CompatibleSslContext) broker.getSslContext(); + assertFalse(sslCtx.getKeyManagers().isEmpty()); + assertFalse(sslCtx.getTrustManagers().isEmpty()); } diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/transport/tcp/TcpFaultyTransportFactory.java b/activemq-unit-tests/src/test/java/org/apache/activemq/transport/tcp/TcpFaultyTransportFactory.java index 592f2fb6f5d..360bd3a7d33 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/transport/tcp/TcpFaultyTransportFactory.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/transport/tcp/TcpFaultyTransportFactory.java @@ -20,17 +20,11 @@ import java.net.URI; import java.net.URISyntaxException; import java.net.UnknownHostException; -import java.util.HashMap; -import java.util.Map; import javax.net.ServerSocketFactory; import javax.net.SocketFactory; import org.apache.activemq.transport.Transport; -import org.apache.activemq.transport.TransportServer; -import org.apache.activemq.util.IOExceptionSupport; -import org.apache.activemq.util.IntrospectionSupport; -import org.apache.activemq.util.URISupport; import org.apache.activemq.wireformat.WireFormat; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -67,22 +61,9 @@ protected TcpFaultyTransportServer createTcpFaultyTransportServer(final URI loca return new TcpFaultyTransportServer(this, location, serverSocketFactory); } - public TransportServer doBind(final URI location) throws IOException { - try { - Map options = new HashMap(URISupport.parseParameters(location)); - - ServerSocketFactory serverSocketFactory = createServerSocketFactory(); - TcpFaultyTransportServer server = createTcpFaultyTransportServer(location, serverSocketFactory); - server.setWireFormatFactory(createWireFormatFactory(options)); - IntrospectionSupport.setProperties(server, options); - Map transportOptions = IntrospectionSupport.extractProperties(options, "transport."); - server.setTransportOption(transportOptions); - server.bind(); - - return server; - } catch (URISyntaxException e) { - throw IOExceptionSupport.create(e); - } + @Override + protected TcpTransportServer createTcpTransportServer(final URI location, ServerSocketFactory serverSocketFactory) throws IOException, URISyntaxException { + return createTcpFaultyTransportServer(location, serverSocketFactory); } diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/NetworkAsyncStartSslTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/NetworkAsyncStartSslTest.java index 0544d503e4b..7c9801eb34a 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/NetworkAsyncStartSslTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/NetworkAsyncStartSslTest.java @@ -39,7 +39,14 @@ public class NetworkAsyncStartSslTest extends JmsMultipleBrokersTestSupport { public static final String TRUST_KEYSTORE = "src/test/resources/client.keystore"; public void testSslPerConnectorConfig() throws Exception { - String transport = "ssl"; + doTestSslPerConnectorConfig("ssl"); + } + + public void testSslPerConnectorConfigNioSsl() throws Exception { + doTestSslPerConnectorConfig("nio+ssl"); + } + + private void doTestSslPerConnectorConfig(String transport) throws Exception { String brokerBUri = transport + "://" + brokerBDomain; String brokerCUri = transport + "://" + brokerCDomain;