Merge pull request #833 from the-thing/proxy_testing

Mina proxy handshake fix and proxy integration tests

Author: chrjohn

quickfixj-core/pom.xml

@@ -117,6 +117,12 @@
 			<version>${slf4j.version}</version>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>io.netty</groupId>
+			<artifactId>netty-example</artifactId>
+			<version>4.1.111.Final</version>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>org.apache.mina</groupId>
 			<artifactId>mina-core</artifactId>
@@ -132,11 +138,11 @@
 			<version>18.3.12</version>
 			<optional>true</optional>
 		</dependency>
-    <dependency>
-      <groupId>com.zaxxer</groupId>
-      <artifactId>HikariCP</artifactId>
-      <version>4.0.3</version>
-    </dependency>
+		<dependency>
+		  <groupId>com.zaxxer</groupId>
+		  <artifactId>HikariCP</artifactId>
+		  <version>4.0.3</version>
+		</dependency>
   </dependencies>
 
 
@@ -218,6 +224,15 @@
 				</configuration>
 			</plugin>
 		</plugins>
+
+	  <extensions>
+		  <!-- required by Netty cross-platform build -->
+		  <extension>
+			  <groupId>kr.motd.maven</groupId>
+			  <artifactId>os-maven-plugin</artifactId>
+			  <version>1.4.0.Final</version>
+		  </extension>
+	  </extensions>
 	</build>
 
 	<reporting>

quickfixj-core/src/main/java/quickfix/mina/CustomSslFilter.java

@@ -0,0 +1,35 @@
+package quickfix.mina;
+
+import org.apache.mina.core.filterchain.IoFilterChain;
+import org.apache.mina.core.session.IoSession;
+import org.apache.mina.filter.ssl.SslFilter;
+
+import javax.net.ssl.SSLContext;
+
+/**
+ * Temporary {@link SslFilter} wrapper that prevents auto connect for initiators.
+ */
+public class CustomSslFilter extends SslFilter {
+
+    private static final boolean DEFAULT_AUTO_START = true;
+
+    private final boolean autoStart;
+
+    public CustomSslFilter(SSLContext sslContext) {
+        this(sslContext, DEFAULT_AUTO_START);
+    }
+
+    public CustomSslFilter(SSLContext sslContext, boolean autoStart) {
+        super(sslContext);
+        this.autoStart = autoStart;
+    }
+
+    @Override
+    public void onPostAdd(IoFilterChain parent, String name, NextFilter next) throws Exception {
+        IoSession session = parent.getSession();
+
+        if (session.isConnected() && autoStart) {
+            onConnected(next, session);
+        }
+    }
+}

quickfixj-core/src/main/java/quickfix/mina/ProtocolFactory.java

@@ -52,24 +52,21 @@ public class ProtocolFactory {
 
     public final static int SOCKET = 0;
     public final static int VM_PIPE = 1;
-    public final static int PROXY = 2;
 
     public static String getTypeString(int type) {
         switch (type) {
         case SOCKET:
             return "SOCKET";
         case VM_PIPE:
             return "VM_PIPE";
-        case PROXY:
-            return "PROXY";
         default:
             return "unknown";
         }
     }
 
     public static SocketAddress createSocketAddress(int transportType, String host,
             int port) throws ConfigError {
-        if (transportType == SOCKET || transportType == PROXY) {
+        if (transportType == SOCKET) {
             return host != null ? new InetSocketAddress(host, port) : new InetSocketAddress(port);
         } else if (transportType == VM_PIPE) {
             return new VmPipeAddress(port);
@@ -94,8 +91,6 @@ public static int getTransportType(String string) {
             return SOCKET;
         } else if (string.equalsIgnoreCase("VM_PIPE")) {
             return VM_PIPE;
-        } else if (string.equalsIgnoreCase("PROXY")) {
-            return PROXY;
         } else {
             throw new RuntimeError("Unknown Transport Type type: " + string);
         }

quickfixj-core/src/main/java/quickfix/mina/acceptor/AbstractSocketAcceptor.java

@@ -40,6 +40,7 @@
 import quickfix.SessionID;
 import quickfix.SessionSettings;
 import quickfix.mina.CompositeIoFilterChainBuilder;
+import quickfix.mina.CustomSslFilter;
 import quickfix.mina.EventHandlingStrategy;
 import quickfix.mina.NetworkingOptions;
 import quickfix.mina.ProtocolFactory;
@@ -134,7 +135,7 @@ private void installSSL(AcceptorSocketDescriptor descriptor,
         log.info("Installing SSL filter for {}", descriptor.getAddress());
         SSLConfig sslConfig = descriptor.getSslConfig();
         SSLContext sslContext = SSLContextFactory.getInstance(sslConfig);
-        SslFilter sslFilter = new SslFilter(sslContext);
+        SslFilter sslFilter = new CustomSslFilter(sslContext);
         sslFilter.setNeedClientAuth(sslConfig.isNeedClientAuth());
         sslFilter.setEnabledCipherSuites(sslConfig.getEnabledCipherSuites() != null ? sslConfig.getEnabledCipherSuites()
                 : SSLSupport.getDefaultCipherSuites(sslContext));

quickfixj-core/src/main/java/quickfix/mina/initiator/IoSessionInitiator.java

@@ -34,6 +34,7 @@
 import quickfix.SessionSettings;
 import quickfix.SystemTime;
 import quickfix.mina.CompositeIoFilterChainBuilder;
+import quickfix.mina.CustomSslFilter;
 import quickfix.mina.EventHandlingStrategy;
 import quickfix.mina.NetworkingOptions;
 import quickfix.mina.ProtocolFactory;
@@ -189,7 +190,7 @@ private void setupIoConnector() throws ConfigError, GeneralSecurityException {
         private SslFilter installSslFilter(CompositeIoFilterChainBuilder ioFilterChainBuilder)
                 throws GeneralSecurityException {
             final SSLContext sslContext = SSLContextFactory.getInstance(sslConfig);
-            final SslFilter sslFilter = new SslFilter(sslContext);
+            final SslFilter sslFilter = new CustomSslFilter(sslContext, false);
             sslFilter.setEnabledCipherSuites(sslConfig.getEnabledCipherSuites() != null ? sslConfig.getEnabledCipherSuites()
                     : SSLSupport.getDefaultCipherSuites(sslContext));
             sslFilter.setEnabledProtocols(sslConfig.getEnabledProtocols() != null ? sslConfig.getEnabledProtocols()

quickfixj-core/src/test/java/quickfix/mina/SocksProxyServer.java

@@ -0,0 +1,74 @@
+package quickfix.mina;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.example.socksproxy.SocksServerInitializer;
+import io.netty.handler.logging.LogLevel;
+import io.netty.handler.logging.LoggingHandler;
+import org.apache.mina.util.DaemonThreadFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ThreadFactory;
+
+/**
+ * Simple SOCKS proxy server based on Netty examples. Only SOCKS protocols are currently supported.
+ * The implementation performs the proxy handshake, but it doesn't perform any user authentication.
+ */
+public class SocksProxyServer {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(SocksProxyServer.class);
+    private static final ThreadFactory THREAD_FACTORY = new DaemonThreadFactory();
+
+    private final ServerBootstrap bootstrap;
+    private final int port;
+    private Channel channel;
+
+    public SocksProxyServer(int port) {
+        this.bootstrap = new ServerBootstrap();
+        this.bootstrap.group(new NioEventLoopGroup(THREAD_FACTORY), new NioEventLoopGroup(THREAD_FACTORY))
+                      .channel(NioServerSocketChannel.class)
+                      .handler(new LoggingHandler(LogLevel.DEBUG))
+                      .childHandler(new SocksServerInitializer());
+        this.port = port;
+    }
+
+    public synchronized void start() {
+        if (channel != null) {
+            throw new IllegalStateException("SOCKS proxy server is running already");
+        }
+
+        try {
+            channel = bootstrap.bind(port)
+                               .sync()
+                               .channel();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException(e);
+        }
+
+        LOGGER.info("SOCKS proxy server started at port: {}", port);
+    }
+
+    public synchronized void stop() {
+        if (channel == null) {
+            throw new IllegalStateException("SOCKS proxy server is not running");
+        }
+
+        try {
+            channel.close().sync();
+            channel = null;
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException("Failed to close SOCKS proxy server");
+        }
+
+        LOGGER.info("SOCKS proxy server stopped at port {}", port);
+    }
+
+    public int getPort() {
+        return port;
+    }
+}