-
Notifications
You must be signed in to change notification settings - Fork 625
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #833 from the-thing/proxy_testing
Mina proxy handshake fix and proxy integration tests (cherry picked from commit 3427f08)
- Loading branch information
Showing
8 changed files
with
374 additions
and
9 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
35 changes: 35 additions & 0 deletions
35
quickfixj-core/src/main/java/quickfix/mina/CustomSslFilter.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
74 changes: 74 additions & 0 deletions
74
quickfixj-core/src/test/java/quickfix/mina/SocksProxyServer.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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; | ||
| } | ||
| } |
177 changes: 177 additions & 0 deletions
177
quickfixj-core/src/test/java/quickfix/mina/SocksProxyTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,177 @@ | ||
| package quickfix.mina; | ||
|
|
||
| import org.apache.mina.util.AvailablePortFinder; | ||
| import org.junit.After; | ||
| import org.junit.Before; | ||
| import org.junit.Test; | ||
| import quickfix.Acceptor; | ||
| import quickfix.ApplicationAdapter; | ||
| import quickfix.ConfigError; | ||
| import quickfix.DefaultMessageFactory; | ||
| import quickfix.FixVersions; | ||
| import quickfix.Initiator; | ||
| import quickfix.MemoryStoreFactory; | ||
| import quickfix.MessageFactory; | ||
| import quickfix.MessageStoreFactory; | ||
| import quickfix.Session; | ||
| import quickfix.SessionFactory; | ||
| import quickfix.SessionID; | ||
| import quickfix.SessionSettings; | ||
| import quickfix.ThreadedSocketAcceptor; | ||
| import quickfix.ThreadedSocketInitiator; | ||
|
|
||
| import java.util.HashMap; | ||
| import java.util.concurrent.TimeUnit; | ||
|
|
||
| /** | ||
| * Performs end to end tests against SOCKS proxy server. | ||
| */ | ||
| public class SocksProxyTest { | ||
|
|
||
| // maximum time to wait for session logon | ||
| private static final long TIMEOUT_SECONDS = 5; | ||
|
|
||
| private SocksProxyServer proxyServer; | ||
|
|
||
| @Before | ||
| public void setUp() { | ||
| int proxyPort = AvailablePortFinder.getNextAvailable(); | ||
|
|
||
| proxyServer = new SocksProxyServer(proxyPort); | ||
| proxyServer.start(); | ||
| } | ||
|
|
||
| @After | ||
| public void tearDown() { | ||
| proxyServer.stop(); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldLoginViaSocks4Proxy() throws ConfigError { | ||
| shouldLoginSocksProxy("4"); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldLoginViaSocks4aProxy() throws ConfigError { | ||
| shouldLoginSocksProxy("4a"); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldLoginViaSocks5Proxy() throws ConfigError { | ||
| shouldLoginSocksProxy("5"); | ||
| } | ||
|
|
||
| private void shouldLoginSocksProxy(String proxyVersion) throws ConfigError { | ||
| int port = AvailablePortFinder.getNextAvailable(); | ||
| SessionConnector acceptor = createAcceptor(port); | ||
|
|
||
| try { | ||
| acceptor.start(); | ||
|
|
||
| SessionConnector initiator = createInitiator(proxyVersion, proxyServer.getPort(), port); | ||
|
|
||
| try { | ||
| initiator.start(); | ||
| assertLoggedOn(acceptor, new SessionID(FixVersions.BEGINSTRING_FIX44, "ALICE", "BOB")); | ||
| assertLoggedOn(initiator, new SessionID(FixVersions.BEGINSTRING_FIX44, "BOB", "ALICE")); | ||
| } finally { | ||
| initiator.stop(); | ||
| } | ||
| } finally { | ||
| acceptor.stop(); | ||
| } | ||
| } | ||
|
|
||
| private void assertLoggedOn(SessionConnector connector, SessionID sessionID) { | ||
| long startTimeNanos = System.nanoTime(); | ||
|
|
||
| while (TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - startTimeNanos) < TIMEOUT_SECONDS) { | ||
| if (isLoggedOn(connector, sessionID)) { | ||
| return; | ||
| } | ||
|
|
||
| try { | ||
| Thread.sleep(100); | ||
| } catch (InterruptedException e) { | ||
| Thread.currentThread().interrupt(); | ||
| throw new RuntimeException("Interrupted", e); | ||
| } | ||
| } | ||
|
|
||
| throw new AssertionError("Session " + sessionID + " is not logged on"); | ||
| } | ||
|
|
||
| private boolean isLoggedOn(SessionConnector connector, SessionID sessionID) { | ||
| Session session = connector.getSessionMap().get(sessionID); | ||
|
|
||
| if (session == null) { | ||
| return false; | ||
| } | ||
|
|
||
| return session.isLoggedOn(); | ||
| } | ||
|
|
||
| private SessionConnector createAcceptor(int port) throws ConfigError { | ||
| MessageStoreFactory messageStoreFactory = new MemoryStoreFactory(); | ||
| MessageFactory messageFactory = new DefaultMessageFactory(); | ||
| SessionSettings acceptorSettings = createAcceptorSettings("ALICE", "BOB", port); | ||
| return new ThreadedSocketAcceptor(new ApplicationAdapter(), messageStoreFactory, acceptorSettings, messageFactory); | ||
| } | ||
|
|
||
| private SessionConnector createInitiator(String proxyVersion, int proxyPort, int port) throws ConfigError { | ||
| MessageStoreFactory messageStoreFactory = new MemoryStoreFactory(); | ||
| MessageFactory messageFactory = new DefaultMessageFactory(); | ||
| SessionSettings initiatorSettings = createInitiatorSettings("BOB", "ALICE", proxyVersion, proxyPort, port); | ||
| return new ThreadedSocketInitiator(new ApplicationAdapter(), messageStoreFactory, initiatorSettings, messageFactory); | ||
| } | ||
|
|
||
| private SessionSettings createAcceptorSettings(String senderId, String targetId, int port) { | ||
| HashMap<Object, Object> defaults = new HashMap<>(); | ||
| defaults.put(SessionFactory.SETTING_CONNECTION_TYPE, "acceptor"); | ||
| defaults.put(Acceptor.SETTING_SOCKET_ACCEPT_PORT, Integer.toString(port)); | ||
| defaults.put(Session.SETTING_START_TIME, "00:00:00"); | ||
| defaults.put(Session.SETTING_END_TIME, "00:00:00"); | ||
| defaults.put(Session.SETTING_HEARTBTINT, "30"); | ||
|
|
||
| SessionID sessionID = new SessionID(FixVersions.BEGINSTRING_FIX44, senderId, targetId); | ||
|
|
||
| SessionSettings sessionSettings = new SessionSettings(); | ||
| sessionSettings.set(defaults); | ||
| sessionSettings.setString(sessionID, "BeginString", FixVersions.BEGINSTRING_FIX44); | ||
| sessionSettings.setString(sessionID, "DataDictionary", "FIX44.xml"); | ||
| sessionSettings.setString(sessionID, "SenderCompID", senderId); | ||
| sessionSettings.setString(sessionID, "TargetCompID", targetId); | ||
|
|
||
| return sessionSettings; | ||
| } | ||
|
|
||
| private SessionSettings createInitiatorSettings(String senderId, String targetId, String proxyVersion, int proxyPort, int port) { | ||
| HashMap<Object, Object> defaults = new HashMap<>(); | ||
| defaults.put(SessionFactory.SETTING_CONNECTION_TYPE, "initiator"); | ||
| defaults.put(Initiator.SETTING_SOCKET_CONNECT_PROTOCOL, ProtocolFactory.getTypeString(ProtocolFactory.SOCKET)); | ||
| defaults.put(Initiator.SETTING_SOCKET_CONNECT_HOST, "localhost"); | ||
| defaults.put(Initiator.SETTING_SOCKET_CONNECT_PORT, Integer.toString(port)); | ||
| defaults.put(Initiator.SETTING_RECONNECT_INTERVAL, "2"); | ||
| defaults.put(Initiator.SETTING_PROXY_HOST, "localhost"); | ||
| defaults.put(Initiator.SETTING_PROXY_PORT, Integer.toString(proxyPort)); | ||
| defaults.put(Initiator.SETTING_PROXY_TYPE, "socks"); | ||
| defaults.put(Initiator.SETTING_PROXY_VERSION, proxyVersion); | ||
| defaults.put(Initiator.SETTING_PROXY_USER, "proxy-user"); | ||
| defaults.put(Initiator.SETTING_PROXY_PASSWORD, "proxy-password"); | ||
| defaults.put(Session.SETTING_START_TIME, "00:00:00"); | ||
| defaults.put(Session.SETTING_END_TIME, "00:00:00"); | ||
| defaults.put(Session.SETTING_HEARTBTINT, "30"); | ||
|
|
||
| SessionID sessionID = new SessionID(FixVersions.BEGINSTRING_FIX44, senderId, targetId); | ||
|
|
||
| SessionSettings sessionSettings = new SessionSettings(); | ||
| sessionSettings.set(defaults); | ||
| sessionSettings.setString(sessionID, "BeginString", FixVersions.BEGINSTRING_FIX44); | ||
| sessionSettings.setString(sessionID, "DataDictionary", "FIX44.xml"); | ||
| sessionSettings.setString(sessionID, "SenderCompID", senderId); | ||
| sessionSettings.setString(sessionID, "TargetCompID", targetId); | ||
|
|
||
| return sessionSettings; | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.