This is an automated email from the ASF dual-hosted git repository. robbie pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/qpid-jms.git
The following commit(s) were added to refs/heads/master by this push: new d06bff9 QPIDJMS-441: add support for supplying proxy handlers to the transports d06bff9 is described below commit d06bff94a089bc1fa36bfbe7e9719777b0a1d61e Author: Stephan Siano <stephan.si...@sap.com> AuthorDate: Wed Oct 9 10:28:02 2019 +0200 QPIDJMS-441: add support for supplying proxy handlers to the transports Changes from Stephan Siano, Robbie Gemmell and Timothy Bish. This closes #32. --- pom.xml | 5 + qpid-jms-client/pom.xml | 5 + .../apache/qpid/jms/JmsConnectionExtensions.java | 22 +- .../qpid/jms/provider/amqp/AmqpProvider.java | 9 + .../qpid/jms/transports/TransportOptions.java | 12 + .../jms/transports/netty/NettyTcpTransport.java | 16 + .../jms/transports/netty/NettyWsTransport.java | 15 +- .../qpid/jms/integration/ProxyIntegrationTest.java | 410 +++++++++++++++++++++ .../org/apache/qpid/jms/test/proxy/TestProxy.java | 389 +++++++++++++++++++ .../qpid/jms/transports/TransportOptionsTest.java | 11 +- .../qpid/jms/transports/netty/NettyServer.java | 10 +- .../transports/netty/NettySimpleAmqpServer.java | 9 +- .../transports/netty/NettySslTransportTest.java | 66 +++- .../transports/netty/NettyTcpToMockServerTest.java | 101 +++-- .../transports/netty/NettyTcpTransportTest.java | 58 +++ .../jms/transports/netty/NettyWsTransportTest.java | 51 +++ .../transports/netty/NettyWssTransportTest.java | 12 + 17 files changed, 1132 insertions(+), 69 deletions(-) diff --git a/pom.xml b/pom.xml index 0745357..6f73d16 100644 --- a/pom.xml +++ b/pom.xml @@ -145,6 +145,11 @@ </dependency> <dependency> <groupId>io.netty</groupId> + <artifactId>netty-handler-proxy</artifactId> + <version>${netty-version}</version> + </dependency> + <dependency> + <groupId>io.netty</groupId> <artifactId>netty-transport</artifactId> <version>${netty-version}</version> </dependency> diff --git a/qpid-jms-client/pom.xml b/qpid-jms-client/pom.xml index b879020..fe3c2b5 100644 --- a/qpid-jms-client/pom.xml +++ b/qpid-jms-client/pom.xml @@ -88,6 +88,11 @@ <artifactId>opentracing-util</artifactId> <scope>provided</scope> </dependency> + <dependency> + <groupId>io.netty</groupId> + <artifactId>netty-handler-proxy</artifactId> + <scope>provided</scope> + </dependency> <!-- =================================== --> <!-- Testing Dependencies --> diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnectionExtensions.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnectionExtensions.java index 46a592f..5e1b03d 100644 --- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnectionExtensions.java +++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsConnectionExtensions.java @@ -19,6 +19,7 @@ package org.apache.qpid.jms; import java.net.URI; import java.util.Map; import java.util.function.BiFunction; +import java.util.function.Supplier; import javax.jms.Connection; import javax.net.ssl.SSLContext; @@ -103,7 +104,26 @@ public enum JmsConnectionExtensions { * </li> * </ul> */ - HTTP_HEADERS_OVERRIDE("httpHeaders"); + HTTP_HEADERS_OVERRIDE("httpHeaders"), + + /** + * Allows a user to inject a custom proxy handler supplier used when creating a transport + * for the connection. + * <p> + * For example, for Netty based transports if a supplier was returned it would provide + * one of Nettys proxy handlers such as HttpProxyHandler, Socks4ProxyHandler, or + * Socks5ProxyHandler created with appropriate login configuration etc. + * <p> + * If the function returns a {@link Supplier}, it must supply a proxy handler when requested. + * <p> + * The extension function takes the form of a BiFunction defined as the following: + * <ul> + * <li> + * {@link BiFunction}<{@link Connection}, {@link URI}, {@link Supplier}> + * </li> + * </ul> + */ + PROXY_HANDLER_SUPPLIER("proxyHandlerSupplier"); private final String extensionKey; diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java index 73a5789..718019b 100644 --- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java +++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpProvider.java @@ -32,6 +32,7 @@ import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; import javax.net.ssl.SSLContext; @@ -204,6 +205,14 @@ public class AmqpProvider implements Provider, TransportListener , AmqpResourceP sslContextOverride = null; } + if (connectionInfo.getExtensionMap().containsKey(JmsConnectionExtensions.PROXY_HANDLER_SUPPLIER)) { + Supplier<?> proxyHandlerSupplier = (Supplier<?>) connectionInfo.getExtensionMap().get( + JmsConnectionExtensions.PROXY_HANDLER_SUPPLIER).apply(connectionInfo.getConnection(), transport.getRemoteLocation()); + if (proxyHandlerSupplier != null) { + transport.getTransportOptions().setProxyHandlerSupplier(proxyHandlerSupplier); + } + } + if (connectionInfo.getExtensionMap().containsKey(JmsConnectionExtensions.HTTP_HEADERS_OVERRIDE)) { @SuppressWarnings({ "unchecked" }) Map<String, String> headers = (Map<String, String>) diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/TransportOptions.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/TransportOptions.java index 2435151..ba62d04 100644 --- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/TransportOptions.java +++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/TransportOptions.java @@ -21,6 +21,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.function.Supplier; import javax.net.ssl.SSLContext; @@ -90,6 +91,7 @@ public class TransportOptions implements Cloneable { private String keyAlias; private int defaultSslPort = DEFAULT_SSL_PORT; private SSLContext sslContextOverride; + private Supplier<?> proxyHandlerSupplier; private final Map<String, String> httpHeaders = new HashMap<>(); @@ -527,6 +529,15 @@ public class TransportOptions implements Cloneable { return sslContextOverride; } + @SuppressWarnings("unchecked") + public <T> Supplier<T> getProxyHandlerSupplier() { + return (Supplier<T>) proxyHandlerSupplier; + } + + public <T> void setProxyHandlerSupplier(Supplier<T> proxyHandlerFactory) { + this.proxyHandlerSupplier = proxyHandlerFactory; + } + // TODO - Expose headers ( ? getWSHeaders : getAuthHeaders ... public Map<String, String> getHttpHeaders() { return httpHeaders; @@ -575,6 +586,7 @@ public class TransportOptions implements Cloneable { copy.setContextProtocol(getContextProtocol()); copy.setDefaultSslPort(getDefaultSslPort()); copy.setSslContextOverride(getSslContextOverride()); + copy.setProxyHandlerSupplier(getProxyHandlerSupplier()); copy.setUseOpenSSL(isUseOpenSSL()); copy.setLocalAddress(getLocalAddress()); copy.setLocalPort(getLocalPort()); diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyTcpTransport.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyTcpTransport.java index 5ddeea4..9f8e826 100644 --- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyTcpTransport.java +++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyTcpTransport.java @@ -19,11 +19,13 @@ package org.apache.qpid.jms.transports.netty; import java.io.IOException; import java.net.URI; import java.security.Principal; +import java.util.Objects; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Supplier; import javax.net.ssl.SSLContext; @@ -51,7 +53,9 @@ import io.netty.channel.SimpleChannelInboundHandler; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.logging.LoggingHandler; +import io.netty.handler.proxy.ProxyHandler; import io.netty.handler.ssl.SslHandler; +import io.netty.resolver.NoopAddressResolverGroup; import io.netty.util.ReferenceCountUtil; import io.netty.util.concurrent.Future; import io.netty.util.concurrent.GenericFutureListener; @@ -464,9 +468,21 @@ public class NettyTcpTransport implements Transport { bootstrap.localAddress(options.getLocalPort()); } } + if (options.getProxyHandlerSupplier() != null) { + // in case we have a proxy we do not want to resolve the address by ourselves but leave this to the proxy + bootstrap.resolver(NoopAddressResolverGroup.INSTANCE); + } } private void configureChannel(final Channel channel) throws Exception { + if (options.getProxyHandlerSupplier() != null) { + Supplier<ProxyHandler> proxyHandlerSupplier = options.getProxyHandlerSupplier(); + + ProxyHandler proxyHandler = proxyHandlerSupplier.get(); + Objects.requireNonNull(proxyHandler, "No proxy handler was returned by the supplier"); + + channel.pipeline().addFirst(proxyHandler); + } if (isSecure()) { final SslHandler sslHandler; try { diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyWsTransport.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyWsTransport.java index 52d4013..8ce8d41 100644 --- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyWsTransport.java +++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/transports/netty/NettyWsTransport.java @@ -32,8 +32,9 @@ import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.channel.ChannelPipeline; import io.netty.handler.codec.http.DefaultHttpHeaders; import io.netty.handler.codec.http.FullHttpResponse; -import io.netty.handler.codec.http.HttpClientCodec; import io.netty.handler.codec.http.HttpObjectAggregator; +import io.netty.handler.codec.http.HttpRequestEncoder; +import io.netty.handler.codec.http.HttpResponseDecoder; import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame; import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame; import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame; @@ -116,7 +117,17 @@ public class NettyWsTransport extends NettyTcpTransport { @Override protected void addAdditionalHandlers(ChannelPipeline pipeline) { - pipeline.addLast(new HttpClientCodec()); + /* + * If we use a HttpClientCodec here instead of the HttpRequestEncoder and the HttpResponseDecoder + * and there is a HttpProxyHandler in the pipeline, that ProxyHandler will add another HttpClientCodec + * for communication with the proxy. When the WebSocketClientHandshaker tries to exchange the codecs in + * the pipeline, it will mix up the two HttpRequestEncoders in the pipeline and exchange the wrong one. + * HttpReqestEncoder and HttpResponseDecoder has precedence over the HttpClientCodec, so the + * WebSocketClientHandshaker will remove these handlers inserted here and will leave the HttpClientCodec + * added by the HttpProxyHandler alone. + */ + pipeline.addLast(new HttpResponseDecoder()); + pipeline.addLast(new HttpRequestEncoder()); pipeline.addLast(new HttpObjectAggregator(8192)); } diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProxyIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProxyIntegrationTest.java new file mode 100644 index 0000000..96fd7b7 --- /dev/null +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProxyIntegrationTest.java @@ -0,0 +1,410 @@ +/* + * 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.qpid.jms.integration; + +import static org.hamcrest.Matchers.equalTo; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.net.InetSocketAddress; +import java.net.Socket; +import java.net.SocketAddress; +import java.net.URI; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; + +import javax.jms.Connection; +import javax.jms.JMSException; +import javax.jms.MessageConsumer; +import javax.jms.Queue; +import javax.jms.Session; +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLSocket; + +import org.apache.qpid.jms.JmsConnection; +import org.apache.qpid.jms.JmsConnectionExtensions; +import org.apache.qpid.jms.JmsConnectionFactory; +import org.apache.qpid.jms.JmsDefaultConnectionListener; +import org.apache.qpid.jms.policy.JmsDefaultPrefetchPolicy; +import org.apache.qpid.jms.test.QpidJmsTestCase; +import org.apache.qpid.jms.test.proxy.TestProxy; +import org.apache.qpid.jms.test.proxy.TestProxy.ProxyType; +import org.apache.qpid.jms.test.testpeer.TestAmqpPeer; +import org.apache.qpid.jms.transports.TransportOptions; +import org.apache.qpid.jms.transports.TransportSupport; +import org.apache.qpid.jms.transports.netty.NettySimpleAmqpServer; +import org.apache.qpid.jms.util.QpidJMSTestRunner; +import org.apache.qpid.jms.util.Repeat; +import org.apache.qpid.proton.amqp.UnsignedInteger; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.netty.handler.proxy.HttpProxyHandler; +import io.netty.handler.proxy.ProxyHandler; +import io.netty.handler.proxy.Socks5ProxyHandler; + +@RunWith(QpidJMSTestRunner.class) +public class ProxyIntegrationTest extends QpidJmsTestCase { + private static final Logger LOG = LoggerFactory.getLogger(ProxyIntegrationTest.class); + private static final String BROKER_JKS_KEYSTORE = "src/test/resources/broker-jks.keystore"; + private static final String CLIENT_JKS_TRUSTSTORE = "src/test/resources/client-jks.truststore"; + private static final String PASSWORD = "password"; + + @Test(timeout = 20000) + public void testCreateConnectionViaSocksProxy() throws Exception { + try (TestAmqpPeer testPeer = new TestAmqpPeer(); + TestProxy testProxy = new TestProxy(ProxyType.SOCKS5)) { + testProxy.start(); + + AtomicInteger supplierUsageCount = new AtomicInteger(); + Supplier<ProxyHandler> proxyHandlerSupplier = () -> { + supplierUsageCount.incrementAndGet(); + return new Socks5ProxyHandler(new InetSocketAddress("localhost", testProxy.getPort())); + }; + + Connection connection = establishConnecton(testPeer, proxyHandlerSupplier, false, null); + + testPeer.expectClose(); + connection.close(); + + assertEquals(1, testProxy.getSuccessCount()); + assertEquals("Unexpected handler supplier usage count", 1, supplierUsageCount.get()); + } + } + + @Test(timeout = 20000) + public void testCreateSecureConnectionViaSocksProxy() throws Exception { + TransportOptions sslOptions = new TransportOptions(); + sslOptions.setKeyStoreLocation(BROKER_JKS_KEYSTORE); + sslOptions.setKeyStorePassword(PASSWORD); + sslOptions.setVerifyHost(false); + + SSLContext context = TransportSupport.createJdkSslContext(sslOptions); + try (TestAmqpPeer testPeer = new TestAmqpPeer(context, false); + TestProxy testProxy = new TestProxy(ProxyType.SOCKS5)) { + testProxy.start(); + + String connOptions = "?transport.trustStoreLocation=" + CLIENT_JKS_TRUSTSTORE + "&" + "transport.trustStorePassword=" + PASSWORD + + "&" + "transport.useOpenSSL=" + false; + AtomicInteger supplierUsageCount = new AtomicInteger(); + Supplier<ProxyHandler> proxyHandlerSupplier = () -> { + supplierUsageCount.incrementAndGet(); + return new Socks5ProxyHandler(new InetSocketAddress("localhost", testProxy.getPort())); + }; + + Connection connection = establishConnecton(testPeer, proxyHandlerSupplier, true, connOptions); + + Socket socket = testPeer.getClientSocket(); + assertTrue(socket instanceof SSLSocket); + + testPeer.expectClose(); + connection.close(); + + assertEquals(1, testProxy.getSuccessCount()); + assertEquals("Unexpected handler supplier usage count", 1, supplierUsageCount.get()); + } + } + + @Repeat(repetitions = 1) + @Test(timeout = 20000) + public void testFailoverCreateConsumerAfterConnectionDropsViaSocksProxy() throws Exception { + try (TestAmqpPeer originalPeer = new TestAmqpPeer(); + TestAmqpPeer finalPeer = new TestAmqpPeer(); + TestProxy testProxy = new TestProxy(ProxyType.SOCKS5)) { + testProxy.start(); + + int proxyPort = testProxy.getPort(); + final CountDownLatch originalConnected = new CountDownLatch(1); + final CountDownLatch finalConnected = new CountDownLatch(1); + + // Create a peer to connect to, then one to reconnect to + final String originalURI = createPeerURI(originalPeer); + final String finalURI = createPeerURI(finalPeer); + + LOG.info("Original peer is at: {}", originalURI); + LOG.info("Final peer is at: {}", finalURI); + LOG.info("Proxy is at port: {}", proxyPort); + + // Connect to the first peer + originalPeer.expectSaslAnonymous(); + originalPeer.expectOpen(); + originalPeer.expectBegin(); + originalPeer.expectBegin(); + originalPeer.dropAfterLastHandler(); + + AtomicInteger supplierUsageCount = new AtomicInteger(); + Supplier<ProxyHandler> proxyHandlerSupplier = () -> { + supplierUsageCount.incrementAndGet(); + return new Socks5ProxyHandler(new InetSocketAddress("localhost", proxyPort)); + }; + + final JmsConnection connection = establishFailoverConnecton(proxyHandlerSupplier, originalPeer, finalPeer); + ((JmsDefaultPrefetchPolicy) connection.getPrefetchPolicy()).setQueuePrefetch(0); + connection.addConnectionListener(new JmsDefaultConnectionListener() { + @Override + public void onConnectionEstablished(URI remoteURI) { + LOG.info("Connection Established: {}", remoteURI); + if (originalURI.equals(remoteURI.toString())) { + originalConnected.countDown(); + } + } + + @Override + public void onConnectionRestored(URI remoteURI) { + LOG.info("Connection Restored: {}", remoteURI); + if (finalURI.equals(remoteURI.toString())) { + finalConnected.countDown(); + } + } + }); + connection.start(); + + assertTrue("Should connect to original peer", originalConnected.await(5, TimeUnit.SECONDS)); + assertEquals("Unexpected handler supplier usage count", 1, supplierUsageCount.get()); + + // --- Post Failover Expectations of FinalPeer --- // + + finalPeer.expectSaslAnonymous(); + finalPeer.expectOpen(); + finalPeer.expectBegin(); + finalPeer.expectBegin(); + finalPeer.expectReceiverAttach(); + finalPeer.expectLinkFlow(false, false, equalTo(UnsignedInteger.valueOf(1))); + finalPeer.expectLinkFlow(true, true, equalTo(UnsignedInteger.valueOf(1))); + finalPeer.expectDetach(true, true, true); + finalPeer.expectClose(); + + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Queue queue = session.createQueue("myQueue"); + MessageConsumer consumer = session.createConsumer(queue); + + assertNull(consumer.receive(500)); + LOG.info("Receive returned"); + + assertTrue("Should connect to final peer", finalConnected.await(5, TimeUnit.SECONDS)); + + LOG.info("Closing consumer"); + consumer.close(); + + // Shut it down + connection.close(); + + finalPeer.waitForAllHandlersToComplete(1000); + + // connection to originalPeer and finalPeer + assertEquals(2, testProxy.getSuccessCount()); + assertEquals("Unexpected handler supplier usage count", 2, supplierUsageCount.get()); + } + } + + @Test(timeout = 30000) + public void testCreateWebSocketConnectionViaHttpProxyAndStart() throws Exception { + doTestCreateWebSocketConnectionViaHttpProxyAndStart(false); + } + + @Test(timeout = 30000) + public void testCreateSecureWebSocketConnectionViaHttpProxyAndStart() throws Exception { + doTestCreateWebSocketConnectionViaHttpProxyAndStart(true); + } + + private void doTestCreateWebSocketConnectionViaHttpProxyAndStart(boolean secure) throws Exception { + TransportOptions options = new TransportOptions(); + options.setKeyStoreLocation(BROKER_JKS_KEYSTORE); + options.setKeyStorePassword(PASSWORD); + options.setVerifyHost(false); + + final AtomicBoolean connectedThroughProxy = new AtomicBoolean(); + + String connOptions = "?transport.trustStoreLocation=" + CLIENT_JKS_TRUSTSTORE + + "&transport.trustStorePassword=" + PASSWORD + + "&transport.useOpenSSL=" + false; + + try (NettySimpleAmqpServer server = new NettySimpleAmqpServer(options, secure, false, true); + TestProxy testProxy = new TestProxy(ProxyType.HTTP)) { + + server.setConnectionIntercepter((protonConnection) -> { + connectedThroughProxy.set(true); + return null; + }); + + server.start(); + testProxy.start(); + + JmsConnectionFactory factory = new JmsConnectionFactory(server.getConnectionURI() + connOptions); + factory.setExtension(JmsConnectionExtensions.PROXY_HANDLER_SUPPLIER.toString(), (connection, remote) -> { + SocketAddress proxyAddress = new InetSocketAddress("localhost", testProxy.getPort()); + Supplier<ProxyHandler> proxyHandlerFactory = () -> { + return new HttpProxyHandler(proxyAddress); + }; + return proxyHandlerFactory; + }); + + Connection connection = factory.createConnection(); + assertNotNull(connection); + + connection.start(); + + Session session = connection.createSession(); + assertNotNull(connection); + session.close(); + + connection.close(); + + assertEquals(1, testProxy.getSuccessCount()); + assertTrue("Client did not connect to test server through the proxy.", connectedThroughProxy.get()); + } + } + + @Test(timeout = 30000) + public void testCreateWebSocketConnectionViaSocksProxyAndStart() throws Exception { + doTestCreateWebSocketConnectionViaSocksProxyAndStart(false); + } + + @Test(timeout = 30000) + public void testCreateSecureWebSocketConnectionViaSocksProxyAndStart() throws Exception { + doTestCreateWebSocketConnectionViaSocksProxyAndStart(true); + } + + private void doTestCreateWebSocketConnectionViaSocksProxyAndStart(boolean secure) throws Exception { + TransportOptions serverOptions = new TransportOptions(); + serverOptions.setKeyStoreLocation(BROKER_JKS_KEYSTORE); + serverOptions.setKeyStorePassword(PASSWORD); + serverOptions.setVerifyHost(false); + + final AtomicBoolean connectedThroughProxy = new AtomicBoolean(); + + String connOptions = "?transport.trustStoreLocation=" + CLIENT_JKS_TRUSTSTORE + + "&transport.trustStorePassword=" + PASSWORD + + "&transport.useOpenSSL=" + false; + + try (NettySimpleAmqpServer server = new NettySimpleAmqpServer(serverOptions, secure, false, true); + TestProxy testProxy = new TestProxy(ProxyType.SOCKS5)) { + + server.setConnectionIntercepter((protonConnection) -> { + connectedThroughProxy.set(true); + return null; + }); + + server.start(); + testProxy.start(); + + JmsConnectionFactory factory = new JmsConnectionFactory(server.getConnectionURI() + connOptions); + factory.setExtension(JmsConnectionExtensions.PROXY_HANDLER_SUPPLIER.toString(), (connection, remote) -> { + SocketAddress proxyAddress = new InetSocketAddress("localhost", testProxy.getPort()); + Supplier<ProxyHandler> proxyHandlerFactory = () -> { + return new Socks5ProxyHandler(proxyAddress); + }; + return proxyHandlerFactory; + }); + + Connection connection = factory.createConnection(); + assertNotNull(connection); + + connection.start(); + + Session session = connection.createSession(); + assertNotNull(connection); + session.close(); + + connection.close(); + + assertEquals(1, testProxy.getSuccessCount()); + assertTrue("Client did not connect to test server through the proxy.", connectedThroughProxy.get()); + } + } + + private Connection establishConnecton(TestAmqpPeer testPeer, Supplier<ProxyHandler> proxyHandlerSupplier, boolean ssl, String optionsString) throws JMSException { + testPeer.expectSaslPlain("guest", "guest"); + testPeer.expectOpen(); + + // Each connection creates a session for managing temporary destinations etc + testPeer.expectBegin(); + + String remoteURI = buildURI(testPeer, ssl, optionsString); + LOG.debug("connect to {}", remoteURI); + JmsConnectionFactory factory = new JmsConnectionFactory(remoteURI); + factory.setExtension(JmsConnectionExtensions.PROXY_HANDLER_SUPPLIER.toString(), (connection1, remote) -> { + return proxyHandlerSupplier; + }); + Connection connection = factory.createConnection("guest", "guest"); + + // Set a clientId to provoke the actual AMQP connection process to occur. + connection.setClientID("clientName"); + + assertNull(testPeer.getThrowable()); + + return connection; + } + + private JmsConnection establishFailoverConnecton(Supplier<ProxyHandler> proxyHandlerSupplier, TestAmqpPeer... peers) throws JMSException { + if (peers.length == 0) { + throw new IllegalArgumentException("No test peers were given, at least 1 required"); + } + + String remoteURI = "failover:("; + boolean first = true; + for (TestAmqpPeer peer : peers) { + if (!first) { + remoteURI += ","; + } + remoteURI += createPeerURI(peer, null); + first = false; + } + + remoteURI += ")?failover.maxReconnectAttempts=10"; + + JmsConnectionFactory factory = new JmsConnectionFactory(remoteURI); + factory.setExtension(JmsConnectionExtensions.PROXY_HANDLER_SUPPLIER.toString(), (connection, remote) -> { + return proxyHandlerSupplier; + }); + + Connection connection = factory.createConnection(); + + return (JmsConnection) connection; + } + + private String createPeerURI(TestAmqpPeer peer) { + return createPeerURI(peer, null); + } + + private String createPeerURI(TestAmqpPeer peer, String params) { + return "amqp://localhost:" + peer.getServerPort() + (params != null ? "?" + params : ""); + } + + private String buildURI(TestAmqpPeer testPeer, boolean ssl, String optionsString) { + String scheme = ssl ? "amqps" : "amqp"; + final String baseURI = scheme + "://localhost:" + testPeer.getServerPort(); + String remoteURI = baseURI; + if (optionsString != null) { + if (optionsString.startsWith("?")) { + remoteURI = baseURI + optionsString; + } else { + remoteURI = baseURI + "?" + optionsString; + } + } + return remoteURI; + } +} diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/test/proxy/TestProxy.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/test/proxy/TestProxy.java new file mode 100644 index 0000000..4bbe8aa --- /dev/null +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/test/proxy/TestProxy.java @@ -0,0 +1,389 @@ +/* + * 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.qpid.jms.test.proxy; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.net.StandardSocketOptions; +import java.nio.ByteBuffer; +import java.nio.channels.AsynchronousServerSocketChannel; +import java.nio.channels.AsynchronousSocketChannel; +import java.nio.channels.Channel; +import java.nio.channels.ClosedChannelException; +import java.nio.channels.CompletionHandler; +import java.nio.charset.StandardCharsets; +import java.util.Objects; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class TestProxy implements AutoCloseable { + private static final Logger LOG = LoggerFactory.getLogger(TestProxy.class); + private static final int TIMEOUT_IN_S = 2; + private AsynchronousServerSocketChannel serverSocketChannel; + private AtomicInteger connectCount = new AtomicInteger(); + private ProxyReadHandler readHandler = new ProxyReadHandler(); + private ProxyWriteHandler writeHandler = new ProxyWriteHandler(); + private int port; + private final ProxyType type; + + public enum ProxyType { + SOCKS5, HTTP + } + + public TestProxy(ProxyType type) throws IOException { + Objects.requireNonNull(type, "Proxy type must be given"); + + this.type = type; + } + + public int getPort() { + return port; + } + + public int getSuccessCount() { + return connectCount.get(); + } + + public void start() throws IOException { + serverSocketChannel = AsynchronousServerSocketChannel.open(); + serverSocketChannel.bind(new InetSocketAddress(0)); + port = ((InetSocketAddress) serverSocketChannel.getLocalAddress()).getPort(); + LOG.info("Bound listen socket to port {}, waiting for clients...", port); + serverSocketChannel.accept(null, new ServerConnectionHandler()); + } + + @Override + public void close() { + LOG.info("stopping proxy server"); + // Close Server Socket + if (serverSocketChannel != null) { + try { + LOG.info("Terminating server socket"); + serverSocketChannel.close(); + } catch (Exception e) { + LOG.error("Cannot close server socket ", e); + } + } + } + + private boolean processHandshakeMessages(ProxyConnectionState attachment) { + if (attachment.handshakePhase == HandshakePhase.INITIAL) { + byte first = attachment.buffer.get(0); + if (first == 0x5) { + attachment.handshakePhase = HandshakePhase.SOCKS5_1; + } else { + attachment.handshakePhase = HandshakePhase.HTTP; + } + } + + if (!assertExpectedHandshakeType(attachment.handshakePhase)) { + LOG.error("Unexpected handshake phase '" + attachment.handshakePhase + "' for proxy of type: " + type); + return false; + } + + switch (attachment.handshakePhase) { + case SOCKS5_1: + return processSocks5Handshake1(attachment); + case SOCKS5_2: + return processSocks5Handshake2(attachment); + case HTTP: + return processHttpHandshake(attachment); + default: + LOG.error("wrong handshake phase"); + return false; + } + } + + private boolean assertExpectedHandshakeType(HandshakePhase handshakePhase) { + switch (handshakePhase) { + case SOCKS5_1: + case SOCKS5_2: + return type == ProxyType.SOCKS5; + case HTTP: + return type == ProxyType.HTTP; + default: + LOG.error("Unknown handshake phase type:" + handshakePhase); + return false; + } + } + + private boolean processHttpHandshake(ProxyConnectionState attachment) { + String requestString = StandardCharsets.ISO_8859_1.decode(attachment.buffer).toString(); + LOG.debug("Request received: {}", requestString); + String requestType = requestString.substring(0, requestString.indexOf(' ')); + String hostandport = requestString.substring(requestType.length() + 1); + hostandport = hostandport.substring(0, hostandport.indexOf(' ')); + String hostname = hostandport.substring(0, hostandport.indexOf(":")); + int port = Integer.parseInt(hostandport.substring(hostname.length() + 1)); + + String line; + if (requestType.equals("CONNECT")) { + LOG.info("CONNECT to {}:{}", hostname, port); + if (connectToServer(hostname, port, attachment)) { + attachment.handshakePhase = HandshakePhase.CONNECTED; + line = "HTTP/1.1 200 Connection established\r\n\r\n"; + } else { + line = "HTTP/1.1 504 Gateway Timeout\r\n\r\n"; + } + } else { + LOG.error("unsupported request type {}", requestType); + line = "HTTP/1.1 502 Bad Gateway\r\n\r\n"; + } + attachment.buffer.clear(); + attachment.buffer.put(StandardCharsets.ISO_8859_1.encode(line)); + attachment.buffer.flip(); + return true; + } + + private boolean processSocks5Handshake1(ProxyConnectionState attachment) { + byte version = attachment.buffer.get(); + if (version != 0x5) { + LOG.error("SOCKS Version {} not supported", version); + closeChannel(attachment.readChannel); + return false; + } + attachment.buffer.clear(); + attachment.buffer.put(version); + attachment.buffer.put((byte) 0x0); // no authentication required + attachment.buffer.flip(); + LOG.info("SOCKS5 connection initialized, no authentication required"); + attachment.handshakePhase = HandshakePhase.SOCKS5_2; + return true; + } + + private boolean processSocks5Handshake2(ProxyConnectionState attachment) { + byte version = attachment.buffer.get(); + if (version != 0x5) { + LOG.error("SOCKS Version {} not supported", version); + closeChannel(attachment.readChannel); + return false; + } + byte command = attachment.buffer.get(); + if (command != 0x1) { + LOG.error("CMD {} not supported", command); + closeChannel(attachment.readChannel); + return false; + } + attachment.buffer.get(); // skip the next byte (should be 0x0) + byte addressType = attachment.buffer.get(); + if (addressType != 0x3) { + LOG.error("Address Type {} not supported", addressType); + closeChannel(attachment.readChannel); + return false; + } + int size = attachment.buffer.get() & 0xFF; // unsigned byte + byte[] hostBytes = new byte[size]; + attachment.buffer.get(hostBytes); + String hostname = new String(hostBytes, StandardCharsets.UTF_8); + int port = attachment.buffer.getShort() & 0xffff; // unsigned short + // now we have hostname and port connect to the actual listen + LOG.info("Create SOCKS5 connection to {}:{}", hostname, port); + if (!connectToServer(hostname, port, attachment)) { + return false; + } + + // write back the client response + attachment.buffer.rewind(); + attachment.buffer.put(1, (byte) 0x0); + attachment.handshakePhase = HandshakePhase.CONNECTED; // handshake done + return true; + } + + private boolean isInHandshake(ProxyConnectionState attachment) { + return attachment.handshakePhase != HandshakePhase.CONNECTED; + } + + private boolean connectToServer(String hostname, int port, ProxyConnectionState attachment) { + try { + AsynchronousSocketChannel serverChannel = AsynchronousSocketChannel.open(); + try { + serverChannel.setOption(StandardSocketOptions.TCP_NODELAY, true); + } catch (IOException e) { + LOG.error("Failed to set TCP_NODELAY before connect, closing channel", e); + closeChannel(serverChannel); + return false; + } + + SocketAddress serverAddr = new InetSocketAddress(hostname, port); + Future<Void> connectResult = serverChannel.connect(serverAddr); + connectResult.get(TIMEOUT_IN_S, TimeUnit.SECONDS); + attachment.writeChannel = serverChannel; + int connectionNumber = connectCount.incrementAndGet(); + LOG.info("Connection {} to {}:{} established", connectionNumber, hostname, port); + ProxyConnectionState serverState = new ProxyConnectionState(); + serverState.readChannel = attachment.writeChannel; + serverState.writeChannel = attachment.readChannel; + serverState.buffer = ByteBuffer.allocate(4096); + serverState.handshakePhase = HandshakePhase.CONNECTED; + + // read from server + serverState.readChannel.read(serverState.buffer, 2, TimeUnit.SECONDS, serverState, readHandler); + return true; + } catch (IOException | InterruptedException | ExecutionException | TimeoutException e) { + LOG.error("connection failed ", e); + closeChannel(attachment.readChannel); + return false; + } + } + + private void closeChannel(Channel channel) { + if (channel != null && channel.isOpen()) { + try { + channel.close(); + } catch (IOException e) { + LOG.error("cannot close", e); + } + } + } + + private void shutdownOutput(AsynchronousSocketChannel channel) { + if (channel != null && channel.isOpen()) { + try { + LOG.info("shutdown output for ({})", channel); + channel.shutdownOutput(); + } catch (IOException e) { + LOG.error("cannot shutdown output to ({})", channel, e); + } + } + } + + private static class ProxyConnectionState { + AsynchronousSocketChannel readChannel; + AsynchronousSocketChannel writeChannel; + ByteBuffer buffer; + HandshakePhase handshakePhase; + } + + private enum HandshakePhase { + INITIAL, HTTP, SOCKS5_1, SOCKS5_2, CONNECTED + } + + private class ServerConnectionHandler implements CompletionHandler<AsynchronousSocketChannel, Object> { + + @Override + public void completed(AsynchronousSocketChannel clientChannel, Object attachment) { + // keep listening + serverSocketChannel.accept(attachment, this); + + ProxyConnectionState clientState = new ProxyConnectionState(); + clientState.readChannel = clientChannel; + clientState.buffer = ByteBuffer.allocate(4096); + clientState.handshakePhase = HandshakePhase.INITIAL; + + try { + clientChannel.setOption(StandardSocketOptions.TCP_NODELAY, true); + } catch (IOException e) { + LOG.error("Failed to set TCP_NODELAY after accept, closing channel", e); + closeChannel(clientChannel); + return; + } + + // read from readChannel + clientChannel.read(clientState.buffer, TIMEOUT_IN_S, TimeUnit.SECONDS, clientState, readHandler); + } + + @Override + public void failed(Throwable e, Object attachment) { + if (!(e instanceof ClosedChannelException)) { + LOG.error("failed to accept connection ", e); + } + closeChannel(serverSocketChannel); + } + } + + private class ProxyReadHandler implements CompletionHandler<Integer, ProxyConnectionState> { + + @Override + public void completed(Integer result, ProxyConnectionState attachment) { + // connection closed + if (result == -1) { + LOG.info("read connection reached end of file ({})", attachment.readChannel); + closeChannel(attachment.readChannel); + shutdownOutput(attachment.writeChannel); + return; + } + LOG.info("read {} bytes (from {})", result, attachment.readChannel); + + attachment.buffer.flip(); + + if (isInHandshake(attachment)) { + if (processHandshakeMessages(attachment)) { + attachment.readChannel.write(attachment.buffer, TIMEOUT_IN_S, TimeUnit.SECONDS, attachment, writeHandler); + } + } else { + // handshake done + if (attachment.writeChannel == null) { + LOG.error("Invalid"); + closeChannel(attachment.readChannel); + return; + } + // forward data to writeChannel + if (attachment.writeChannel.isOpen()) { + attachment.writeChannel.write(attachment.buffer, TIMEOUT_IN_S, TimeUnit.SECONDS, attachment, writeHandler); + } else { + closeChannel(attachment.readChannel); + } + } + } + + @Override + public void failed(Throwable e, ProxyConnectionState attachment) { + if (!(e instanceof ClosedChannelException)) { + LOG.info("read failed", e); + } + closeChannel(attachment.writeChannel); + closeChannel(attachment.readChannel); + } + } + + private class ProxyWriteHandler implements CompletionHandler<Integer, ProxyConnectionState> { + + @Override + public void completed(Integer result, ProxyConnectionState attachment) { + // connection closed + if (result == -1) { + LOG.info("write connection closed"); + closeChannel(attachment.writeChannel); + closeChannel(attachment.readChannel); + return; + } + LOG.debug("wrote {} bytes", result); + attachment.buffer.clear(); + if (attachment.readChannel.isOpen()) { + attachment.readChannel.read(attachment.buffer, attachment, readHandler); + } else { + closeChannel(attachment.writeChannel); + } + } + + @Override + public void failed(Throwable e, ProxyConnectionState attachment) { + if (!(e instanceof ClosedChannelException)) { + LOG.info("write failed", e); + } + closeChannel(attachment.writeChannel); + closeChannel(attachment.readChannel); + } + } +} diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/TransportOptionsTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/TransportOptionsTest.java index 272cd86..402608b 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/TransportOptionsTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/TransportOptionsTest.java @@ -22,12 +22,16 @@ import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertNull; import static org.junit.Assert.fail; +import java.util.function.Supplier; + import javax.net.ssl.SSLContext; import org.apache.qpid.jms.test.QpidJmsTestCase; import org.junit.Test; import org.mockito.Mockito; +import io.netty.handler.proxy.ProxyHandler; + /** * Test for class TransportOptions */ @@ -64,13 +68,13 @@ public class TransportOptionsTest extends QpidJmsTestCase { private static final String[] DISABLED_CIPHERS = new String[] {"CIPHER_C"}; private static final SSLContext SSL_CONTEXT = Mockito.mock(SSLContext.class); + private static final Supplier<ProxyHandler> PROXY_HANDLER_SUPPLIER = () -> Mockito.mock(ProxyHandler.class); private static final String JAVAX_NET_SSL_KEY_STORE = "javax.net.ssl.keyStore"; private static final String JAVAX_NET_SSL_KEY_STORE_PASSWORD = "javax.net.ssl.keyStorePassword"; private static final String JAVAX_NET_SSL_TRUST_STORE = "javax.net.ssl.trustStore"; private static final String JAVAX_NET_SSL_TRUST_STORE_PASSWORD = "javax.net.ssl.trustStorePassword"; - @Test public void testCreate() { TransportOptions options = new TransportOptions(); @@ -93,6 +97,7 @@ public class TransportOptionsTest extends QpidJmsTestCase { assertNull(options.getTrustStorePassword()); assertNull(options.getKeyAlias()); assertNull(options.getSslContextOverride()); + assertNull(options.getProxyHandlerSupplier()); } @Test @@ -139,6 +144,7 @@ public class TransportOptionsTest extends QpidJmsTestCase { assertEquals(LOCAL_ADDRESS,options.getLocalAddress()); assertEquals(LOCAL_PORT,options.getLocalPort()); assertEquals(SSL_CONTEXT, options.getSslContextOverride()); + assertEquals(PROXY_HANDLER_SUPPLIER, options.getProxyHandlerSupplier()); assertArrayEquals(ENABLED_PROTOCOLS,options.getEnabledProtocols()); assertArrayEquals(DISABLED_PROTOCOLS,options.getDisabledProtocols()); assertArrayEquals(ENABLED_CIPHERS,options.getEnabledCipherSuites()); @@ -220,6 +226,7 @@ public class TransportOptionsTest extends QpidJmsTestCase { assertEquals(KEY_ALIAS, options.getKeyAlias()); assertEquals(CONTEXT_PROTOCOL, options.getContextProtocol()); assertEquals(SSL_CONTEXT, options.getSslContextOverride()); + assertEquals(PROXY_HANDLER_SUPPLIER, options.getProxyHandlerSupplier()); assertArrayEquals(ENABLED_PROTOCOLS,options.getEnabledProtocols()); assertArrayEquals(DISABLED_PROTOCOLS,options.getDisabledProtocols()); assertArrayEquals(ENABLED_CIPHERS,options.getEnabledCipherSuites()); @@ -253,6 +260,7 @@ public class TransportOptionsTest extends QpidJmsTestCase { options.setConnectTimeout(TEST_CONNECT_TIMEOUT); options.setDefaultSslPort(TEST_DEFAULT_SSL_PORT); options.setSslContextOverride(SSL_CONTEXT); + options.setProxyHandlerSupplier(PROXY_HANDLER_SUPPLIER); return options; } @@ -328,6 +336,7 @@ public class TransportOptionsTest extends QpidJmsTestCase { options.setDisabledCipherSuites(DISABLED_CIPHERS); options.setLocalAddress(LOCAL_ADDRESS); options.setLocalPort(LOCAL_PORT); + options.setProxyHandlerSupplier(PROXY_HANDLER_SUPPLIER); return options; } diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyServer.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyServer.java index 2046854..b22e916 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyServer.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyServer.java @@ -25,6 +25,7 @@ import java.nio.charset.StandardCharsets; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import javax.net.ssl.SSLContext; import javax.net.ssl.SSLEngine; @@ -92,6 +93,7 @@ public abstract class NettyServer implements AutoCloseable { private volatile SslHandler sslHandler; private volatile HandshakeComplete handshakeComplete; private final CountDownLatch handshakeCompletion = new CountDownLatch(1); + private final AtomicInteger channelActiveCount = new AtomicInteger(); private final AtomicBoolean started = new AtomicBoolean(); @@ -154,7 +156,7 @@ public abstract class NettyServer implements AutoCloseable { return handshakeComplete; } - protected URI getConnectionURI() throws Exception { + public URI getConnectionURI() throws Exception { if (!started.get()) { throw new IllegalStateException("Cannot get URI of non-started server"); } @@ -326,6 +328,8 @@ public abstract class NettyServer implements AutoCloseable { } }); } + + channelActiveCount.incrementAndGet(); } @Override @@ -383,4 +387,8 @@ public abstract class NettyServer implements AutoCloseable { protected SslHandler getSslHandler() { return sslHandler; } + + public int getChannelActiveCount() { + return channelActiveCount.get(); + } } diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySimpleAmqpServer.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySimpleAmqpServer.java index 01d1f32..e5054d0 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySimpleAmqpServer.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySimpleAmqpServer.java @@ -37,6 +37,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadFactory; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; import org.apache.qpid.jms.transports.TransportOptions; import org.apache.qpid.jms.util.IdGenerator; @@ -284,7 +285,7 @@ public class NettySimpleAmqpServer extends NettyServer { protonConnection.open(); if (connectionIntercepter != null) { - failure = connectionIntercepter.interceptConnectionAttempt(connection); + failure = connectionIntercepter.apply(connection); } if (failure == null) { @@ -496,9 +497,11 @@ public class NettySimpleAmqpServer extends NettyServer { this.connectionIntercepter = connectionIntercepter; } - public interface ConnectionIntercepter { + @FunctionalInterface + public interface ConnectionIntercepter extends Function<Connection, ErrorCondition> { - ErrorCondition interceptConnectionAttempt(Connection connection); + @Override + ErrorCondition apply(Connection connection); } diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySslTransportTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySslTransportTest.java index 3a562cc..1205587 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySslTransportTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettySslTransportTest.java @@ -23,12 +23,17 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.SocketAddress; import java.net.URI; import java.net.URISyntaxException; import java.security.cert.Certificate; import java.security.cert.X509Certificate; import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; +import org.apache.qpid.jms.test.Wait; +import org.apache.qpid.jms.test.proxy.TestProxy; import org.apache.qpid.jms.transports.Transport; import org.apache.qpid.jms.transports.TransportListener; import org.apache.qpid.jms.transports.TransportOptions; @@ -39,6 +44,10 @@ import org.junit.runner.RunWith; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import io.netty.handler.proxy.HttpProxyHandler; +import io.netty.handler.proxy.ProxyHandler; +import io.netty.handler.proxy.Socks5ProxyHandler; + /** * Test basic functionality of the Netty based TCP Transport ruuing in secure mode (SSL). */ @@ -238,19 +247,34 @@ public class NettySslTransportTest extends NettyTcpTransportTest { @Test(timeout = 60 * 1000) public void testConnectToServerVerifyHost() throws Exception { - doConnectToServerVerifyHostTestImpl(true); + doConnectToServerVerifyHostTestImpl(true, null); } @Test(timeout = 60 * 1000) public void testConnectToServerNoVerifyHost() throws Exception { - doConnectToServerVerifyHostTestImpl(false); + doConnectToServerVerifyHostTestImpl(false, null); } - private void doConnectToServerVerifyHostTestImpl(boolean verifyHost) throws Exception, URISyntaxException, IOException, InterruptedException { + @Test(timeout = 60 * 1000) + public void testConnectViaSocksProxyToServerVerifyHost() throws Exception { + doConnectToServerVerifyHostTestImpl(true, TestProxy.ProxyType.SOCKS5); + } + + @Test(timeout = 60 * 1000) + public void testConnectViaSocksProxyToServerNoVerifyHost() throws Exception { + doConnectToServerVerifyHostTestImpl(false, TestProxy.ProxyType.SOCKS5); + } + + protected void doConnectToServerVerifyHostTestImpl(boolean verifyHost, TestProxy.ProxyType proxyType) throws Exception { TransportOptions serverOptions = createServerOptions(); serverOptions.setKeyStoreLocation(SERVER_WRONG_HOST_KEYSTORE); + TestProxy testProxy = null; try (NettyEchoServer server = createEchoServer(serverOptions)) { + if (proxyType != null) { + testProxy = new TestProxy(proxyType); + testProxy.start(); + } server.start(); int port = server.getServerPort(); @@ -258,6 +282,10 @@ public class NettySslTransportTest extends NettyTcpTransportTest { TransportOptions clientOptions = createClientOptionsIsVerify(verifyHost); + if (proxyType != null) { + configureProxyHandlerSupplier(proxyType, testProxy, clientOptions); + } + if (verifyHost) { assertTrue("Expected verifyHost to be true", clientOptions.isVerifyHost()); } else { @@ -286,6 +314,38 @@ public class NettySslTransportTest extends NettyTcpTransportTest { } transport.close(); + + if (proxyType != null) { + assertEquals(1, testProxy.getSuccessCount()); + } + assertTrue(Wait.waitFor(new Wait.Condition() { + @Override + public boolean isSatisfied() throws Exception { + return server.getChannelActiveCount() == 1; + } + }, 10_000, 10)); + } finally { + if (testProxy != null) { + testProxy.close(); + } + } + } + + protected void configureProxyHandlerSupplier(TestProxy.ProxyType proxyType, TestProxy testProxy, TransportOptions clientOptions) { + if (proxyType == TestProxy.ProxyType.SOCKS5) { + SocketAddress proxyAddress = new InetSocketAddress("localhost", testProxy.getPort()); + Supplier<ProxyHandler> proxyHandlerFactory = () -> { + return new Socks5ProxyHandler(proxyAddress); + }; + clientOptions.setProxyHandlerSupplier(proxyHandlerFactory); + } else if (proxyType == TestProxy.ProxyType.HTTP) { + SocketAddress proxyAddress = new InetSocketAddress("localhost", testProxy.getPort()); + Supplier<ProxyHandler> proxyHandlerFactory = () -> { + return new HttpProxyHandler(proxyAddress); + }; + clientOptions.setProxyHandlerSupplier(proxyHandlerFactory); + } else { + throw new IllegalArgumentException("Unknown proxy type:" + proxyType); } } diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyTcpToMockServerTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyTcpToMockServerTest.java index b9f5b07..ea067bb 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyTcpToMockServerTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyTcpToMockServerTest.java @@ -41,7 +41,6 @@ import org.apache.qpid.jms.JmsConnectionExtensions; import org.apache.qpid.jms.JmsConnectionFactory; import org.apache.qpid.jms.test.QpidJmsTestCase; import org.apache.qpid.jms.transports.TransportOptions; -import org.apache.qpid.jms.transports.netty.NettySimpleAmqpServer.ConnectionIntercepter; import org.apache.qpid.proton.amqp.Symbol; import org.apache.qpid.proton.amqp.transport.ConnectionError; import org.apache.qpid.proton.amqp.transport.ErrorCondition; @@ -159,38 +158,31 @@ public class NettyTcpToMockServerTest extends QpidJmsTestCase { final CountDownLatch redirectComplete = new CountDownLatch(1); - server.setConnectionIntercepter(new ConnectionIntercepter() { + server.setConnectionIntercepter((protonConnection) -> { - @Override - public ErrorCondition interceptConnectionAttempt(org.apache.qpid.proton.engine.Connection connection) { - ErrorCondition redirection = new ErrorCondition(ConnectionError.REDIRECT, - "Server redirecting connection."); + ErrorCondition redirection = new ErrorCondition(ConnectionError.REDIRECT, + "Server redirecting connection."); - URI serverURI = null; - try { - serverURI = redirect.getConnectionURI(); - } catch (Exception e) { - new RuntimeException(); - } + URI serverURI = null; + try { + serverURI = redirect.getConnectionURI(); + } catch (Exception e) { + new RuntimeException(); + } - // Create standard redirection condition - Map<Symbol, Object> infoMap = new HashMap<Symbol, Object> (); - infoMap.put(OPEN_HOSTNAME, serverURI.getHost()); - infoMap.put(NETWORK_HOST, serverURI.getHost()); - infoMap.put(PORT, serverURI.getPort()); - redirection.setInfo(infoMap); + // Create standard redirection condition + Map<Symbol, Object> infoMap = new HashMap<Symbol, Object> (); + infoMap.put(OPEN_HOSTNAME, serverURI.getHost()); + infoMap.put(NETWORK_HOST, serverURI.getHost()); + infoMap.put(PORT, serverURI.getPort()); + redirection.setInfo(infoMap); - return redirection; - } + return redirection; }); - redirect.setConnectionIntercepter(new ConnectionIntercepter() { - - @Override - public ErrorCondition interceptConnectionAttempt(org.apache.qpid.proton.engine.Connection connection) { - redirectComplete.countDown(); - return null; - } + redirect.setConnectionIntercepter((protonConnection) -> { + redirectComplete.countDown(); + return null; }); server.start(); @@ -234,40 +226,33 @@ public class NettyTcpToMockServerTest extends QpidJmsTestCase { primary.setWebSocketPath("/primary"); redirect.setWebSocketPath("/redirect"); - primary.setConnectionIntercepter(new ConnectionIntercepter() { - - @Override - public ErrorCondition interceptConnectionAttempt(org.apache.qpid.proton.engine.Connection connection) { - ErrorCondition redirection = new ErrorCondition(ConnectionError.REDIRECT, - "Server redirecting connection."); - - URI serverURI = null; - try { - serverURI = redirect.getConnectionURI(); - } catch (Exception e) { - new RuntimeException(); - } - - // Create standard redirection condition - Map<Symbol, Object> infoMap = new HashMap<Symbol, Object> (); - infoMap.put(OPEN_HOSTNAME, serverURI.getHost()); - infoMap.put(NETWORK_HOST, serverURI.getHost()); - infoMap.put(PORT, serverURI.getPort()); - infoMap.put(PATH, redirect.getWebSocketPath()); - infoMap.put(SCHEME, redirect.isSecureServer() ? "wss" : "ws"); - redirection.setInfo(infoMap); - - return redirection; - } - }); + primary.setConnectionIntercepter((protonConnection) -> { - redirect.setConnectionIntercepter(new ConnectionIntercepter() { + ErrorCondition redirection = new ErrorCondition(ConnectionError.REDIRECT, + "Server redirecting connection."); - @Override - public ErrorCondition interceptConnectionAttempt(org.apache.qpid.proton.engine.Connection connection) { - redirectComplete.countDown(); - return null; + URI serverURI = null; + try { + serverURI = redirect.getConnectionURI(); + } catch (Exception e) { + new RuntimeException(); } + + // Create standard redirection condition + Map<Symbol, Object> infoMap = new HashMap<Symbol, Object> (); + infoMap.put(OPEN_HOSTNAME, serverURI.getHost()); + infoMap.put(NETWORK_HOST, serverURI.getHost()); + infoMap.put(PORT, serverURI.getPort()); + infoMap.put(PATH, redirect.getWebSocketPath()); + infoMap.put(SCHEME, redirect.isSecureServer() ? "wss" : "ws"); + redirection.setInfo(infoMap); + + return redirection; + }); + + redirect.setConnectionIntercepter((protonConnection) -> { + redirectComplete.countDown(); + return null; }); primary.start(); diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyTcpTransportTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyTcpTransportTest.java index 87cf074..74ce189 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyTcpTransportTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyTcpTransportTest.java @@ -27,14 +27,19 @@ import static org.junit.Assume.assumeTrue; import java.io.IOException; import java.lang.reflect.Field; +import java.net.InetSocketAddress; +import java.net.SocketAddress; import java.net.URI; import java.util.ArrayList; import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; import org.apache.qpid.jms.test.QpidJmsTestCase; import org.apache.qpid.jms.test.Wait; +import org.apache.qpid.jms.test.proxy.TestProxy; +import org.apache.qpid.jms.test.proxy.TestProxy.ProxyType; import org.apache.qpid.jms.transports.Transport; import org.apache.qpid.jms.transports.TransportListener; import org.apache.qpid.jms.transports.TransportOptions; @@ -53,6 +58,8 @@ import io.netty.channel.epoll.Epoll; import io.netty.channel.epoll.EpollEventLoopGroup; import io.netty.channel.kqueue.KQueue; import io.netty.channel.kqueue.KQueueEventLoopGroup; +import io.netty.handler.proxy.ProxyHandler; +import io.netty.handler.proxy.Socks5ProxyHandler; import io.netty.util.ResourceLeakDetector; import io.netty.util.ResourceLeakDetector.Level; @@ -617,6 +624,57 @@ public class NettyTcpTransportTest extends QpidJmsTestCase { doTestEpollSupport(false); } + @Test(timeout = 60 * 1000) + public void testConnectToServerViaProxy() throws Exception { + try (TestProxy testProxy = new TestProxy(ProxyType.SOCKS5); + NettyEchoServer server = createEchoServer(createServerOptions())) { + + testProxy.start(); + server.start(); + + int port = server.getServerPort(); + LOG.info("Echo server bound at: {}", port); + URI serverLocation = new URI("tcp://localhost:" + port); + + TransportOptions clientOptions = createClientOptions(); + SocketAddress proxyAddress = new InetSocketAddress("localhost", testProxy.getPort()); + Supplier<ProxyHandler> proxyHandlerFactory = () -> { + return new Socks5ProxyHandler(proxyAddress); + }; + clientOptions.setProxyHandlerSupplier(proxyHandlerFactory); + + Transport transport = createTransport(serverLocation, testListener, clientOptions); + try { + transport.connect(null, null); + LOG.info("Connected to server:{} as expected.", serverLocation); + } catch (Exception e) { + fail("Should have connected to the server at " + serverLocation + " but got exception: " + e); + } + + assertTrue(transport.isConnected()); + assertEquals(serverLocation, transport.getRemoteLocation()); + + assertTrue(Wait.waitFor(new Wait.Condition() { + @Override + public boolean isSatisfied() throws Exception { + return server.getChannelActiveCount() == 1; + } + }, 10_000, 10)); + + assertEquals(1, testProxy.getSuccessCount()); + + transport.close(); + + // Additional close should not fail or cause other problems. + transport.close(); + + } + + assertTrue(!transportClosed); // Normal shutdown does not trigger the event. + assertTrue(exceptions.isEmpty()); + assertTrue(data.isEmpty()); + } + private void doTestEpollSupport(boolean useEpoll) throws Exception { assumeTrue(Epoll.isAvailable()); diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyWsTransportTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyWsTransportTest.java index 4e3164c..655be3f 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyWsTransportTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyWsTransportTest.java @@ -23,11 +23,16 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.SocketAddress; import java.net.URI; import java.util.ArrayList; import java.util.List; +import java.util.function.Supplier; import org.apache.qpid.jms.test.Wait; +import org.apache.qpid.jms.test.proxy.TestProxy; +import org.apache.qpid.jms.test.proxy.TestProxy.ProxyType; import org.apache.qpid.jms.transports.Transport; import org.apache.qpid.jms.transports.TransportListener; import org.apache.qpid.jms.transports.TransportOptions; @@ -40,6 +45,8 @@ import io.netty.buffer.ByteBufUtil; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.HttpHeaders; import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler.HandshakeComplete; +import io.netty.handler.proxy.HttpProxyHandler; +import io.netty.handler.proxy.ProxyHandler; /** * Test the Netty based WebSocket Transport @@ -387,4 +394,48 @@ public class NettyWsTransportTest extends NettyTcpTransportTest { assertTrue(exceptions.isEmpty()); assertTrue(data.isEmpty()); } + + @Test(timeout = 60000) + public void testConnectViaHttpProxy() throws Exception { + try (TestProxy testProxy = new TestProxy(ProxyType.HTTP); + NettyEchoServer server = createEchoServer(createServerOptions())) { + testProxy.start(); + server.start(); + + int port = server.getServerPort(); + URI serverLocation = new URI("tcp://localhost:" + port); + + TransportOptions clientOptions = createClientOptions(); + SocketAddress proxyAddress = new InetSocketAddress("localhost", testProxy.getPort()); + Supplier<ProxyHandler> proxyHandlerFactory = () -> { + return new HttpProxyHandler(proxyAddress); + }; + clientOptions.setProxyHandlerSupplier(proxyHandlerFactory); + + Transport transport = createTransport(serverLocation, testListener, clientOptions); + try { + transport.connect(null, null); + LOG.info("Connected to server:{} as expected.", serverLocation); + } catch (Exception e) { + fail("Should have connected to the server at " + serverLocation + " but got exception: " + e); + } + + assertTrue(transport.isConnected()); + assertEquals(serverLocation, transport.getRemoteLocation()); + + transport.close(); + + assertEquals(1, testProxy.getSuccessCount()); + assertTrue(Wait.waitFor(new Wait.Condition() { + @Override + public boolean isSatisfied() throws Exception { + return server.getChannelActiveCount() == 1; + } + }, 10_000, 10)); + } + + assertTrue(!transportClosed); // Normal shutdown does not trigger the event. + assertTrue(exceptions.isEmpty()); + assertTrue(data.isEmpty()); + } } diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyWssTransportTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyWssTransportTest.java index f157537..1f2a4bd 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyWssTransportTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/transports/netty/NettyWssTransportTest.java @@ -18,8 +18,10 @@ package org.apache.qpid.jms.transports.netty; import java.net.URI; +import org.apache.qpid.jms.test.proxy.TestProxy; import org.apache.qpid.jms.transports.TransportListener; import org.apache.qpid.jms.transports.TransportOptions; +import org.junit.Test; /** * Test the NettyWsTransport with channel level security enabled. @@ -39,4 +41,14 @@ public class NettyWssTransportTest extends NettySslTransportTest { return new NettyWsTransport(listener, serverLocation, options, true); } } + + @Test(timeout = 60 * 1000) + public void testConnectViaHttpProxyToServerVerifyHost() throws Exception { + doConnectToServerVerifyHostTestImpl(true, TestProxy.ProxyType.HTTP); + } + + @Test(timeout = 60 * 1000) + public void testConnectViaHttpProxyToServerNoVerifyHost() throws Exception { + doConnectToServerVerifyHostTestImpl(false, TestProxy.ProxyType.HTTP); + } } --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@qpid.apache.org For additional commands, e-mail: commits-h...@qpid.apache.org