diff options
Diffstat (limited to 'java/client/src')
8 files changed, 191 insertions, 74 deletions
diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java b/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java index 8fb87f9e64..5492bd91e7 100644 --- a/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java +++ b/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java @@ -35,6 +35,8 @@ public class AMQBrokerDetails implements BrokerDetails private String _transport; private HashMap<String, String> _options; + + private SSLConfiguration _sslConfiguration; public AMQBrokerDetails() { @@ -174,15 +176,11 @@ public class AMQBrokerDetails implements BrokerDetails } } - public AMQBrokerDetails(String host, int port, boolean useSSL) + public AMQBrokerDetails(String host, int port, SSLConfiguration sslConfiguration) { _host = host; _port = port; - - if (useSSL) - { - setOption(OPTIONS_SSL, "true"); - } + _sslConfiguration = sslConfiguration; } public String getHost() @@ -247,6 +245,16 @@ public class AMQBrokerDetails implements BrokerDetails { setOption(OPTIONS_CONNECT_TIMEOUT, Long.toString(timeout)); } + + public SSLConfiguration getSSLConfiguration() + { + return _sslConfiguration; + } + + public void setSSLConfiguration(SSLConfiguration sslConfig) + { + _sslConfiguration = sslConfig; + } public String toString() { @@ -280,8 +288,7 @@ public class AMQBrokerDetails implements BrokerDetails return _host.equalsIgnoreCase(bd.getHost()) && (_port == bd.getPort()) && _transport.equalsIgnoreCase(bd.getTransport()) && - (useSSL() == bd.useSSL()); - + compareSSLConfigurations(bd.getSSLConfiguration()); //todo do we need to compare all the options as well? } @@ -313,27 +320,25 @@ public class AMQBrokerDetails implements BrokerDetails return optionsURL.toString(); } - - public boolean useSSL() + + // Do we need to do a more in-depth comparison? + private boolean compareSSLConfigurations(SSLConfiguration other) { - // To be friendly to users we should be case insensitive. - // or simply force users to conform to OPTIONS_SSL - // todo make case insensitive by trying ssl Ssl sSl ssL SSl SsL sSL SSL - - if (_options.containsKey(OPTIONS_SSL)) - { - return _options.get(OPTIONS_SSL).equalsIgnoreCase("true"); - } - - return USE_SSL_DEFAULT; + boolean retval = false; + if (_sslConfiguration == null && + other == null) + { + retval = true; + } + else if (_sslConfiguration != null && + other != null) + { + retval = true; + } + + return retval; } - public void useSSL(boolean ssl) - { - setOption(OPTIONS_SSL, Boolean.toString(ssl)); - } - - public static String checkTransport(String broker) { if ((!broker.contains("://"))) diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java b/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java index 50299fa9d5..fbf9e3c4f7 100644 --- a/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java +++ b/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java @@ -141,6 +141,11 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect * The connection meta data */ private QpidConnectionMetaData _connectionMetaData; + + /** + * Configuration info for SSL + */ + private SSLConfiguration _sslConfiguration; /** * @param broker brokerdetails @@ -157,17 +162,43 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect this(new AMQConnectionURL(ConnectionURL.AMQ_PROTOCOL + "://" + username + ":" + password + "@" + (clientName == null ? "" : clientName) + "/" + - virtualHost + "?brokerlist='" + AMQBrokerDetails.checkTransport(broker) + "'")); + virtualHost + "?brokerlist='" + AMQBrokerDetails.checkTransport(broker) + "'"), null); } + + /** + * @param broker brokerdetails + * @param username username + * @param password password + * @param clientName clientid + * @param virtualHost virtualhost + * @throws AMQException + * @throws URLSyntaxException + */ + public AMQConnection(String broker, String username, String password, + String clientName, String virtualHost, SSLConfiguration sslConfig) throws AMQException, URLSyntaxException + { + this(new AMQConnectionURL(ConnectionURL.AMQ_PROTOCOL + "://" + + username + ":" + password + "@" + + (clientName == null ? "" : clientName) + "/" + + virtualHost + "?brokerlist='" + AMQBrokerDetails.checkTransport(broker) + "'"), sslConfig); + } + public AMQConnection(String host, int port, String username, String password, String clientName, String virtualHost) throws AMQException, URLSyntaxException { - this(host, port, false, username, password, clientName, virtualHost); + this(host, port, false, username, password, clientName, virtualHost, null); } + + public AMQConnection(String host, int port, String username, String password, + String clientName, String virtualHost, SSLConfiguration sslConfig) throws AMQException, URLSyntaxException + { + this(host, port, false, username, password, clientName, virtualHost, sslConfig); + } + public AMQConnection(String host, int port, boolean useSSL, String username, String password, - String clientName, String virtualHost) throws AMQException, URLSyntaxException + String clientName, String virtualHost, SSLConfiguration sslConfig) throws AMQException, URLSyntaxException { this(new AMQConnectionURL(useSSL ? ConnectionURL.AMQ_PROTOCOL + "://" + @@ -180,18 +211,24 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect (clientName == null ? "" : clientName) + virtualHost + "?brokerlist='tcp://" + host + ":" + port + "'" + "," + ConnectionURL.OPTIONS_SSL + "='false'" - )); + ), sslConfig); } public AMQConnection(String connection) throws AMQException, URLSyntaxException { - this(new AMQConnectionURL(connection)); + this(new AMQConnectionURL(connection), null); } + + public AMQConnection(String connection, SSLConfiguration sslConfig) throws AMQException, URLSyntaxException + { + this(new AMQConnectionURL(connection), sslConfig); + } + - public AMQConnection(ConnectionURL connectionURL) throws AMQException + public AMQConnection(ConnectionURL connectionURL, SSLConfiguration sslConfig) throws AMQException { _logger.info("Connection:" + connectionURL); - + _sslConfiguration = sslConfig; if (connectionURL == null) { throw new IllegalArgumentException("Connection must be specified"); @@ -319,9 +356,9 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect } } - public boolean attemptReconnection(String host, int port, boolean useSSL) + public boolean attemptReconnection(String host, int port) { - BrokerDetails bd = new AMQBrokerDetails(host, port, useSSL); + BrokerDetails bd = new AMQBrokerDetails(host, port, _sslConfiguration); _failoverPolicy.setBroker(bd); @@ -1017,5 +1054,9 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect AMQConnectionFactory.class.getName(), null); // factory location } - + + public SSLConfiguration getSSLConfiguration() + { + return _sslConfiguration; + } } diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQConnectionFactory.java b/java/client/src/main/java/org/apache/qpid/client/AMQConnectionFactory.java index 17af3702a4..608f0a028d 100644 --- a/java/client/src/main/java/org/apache/qpid/client/AMQConnectionFactory.java +++ b/java/client/src/main/java/org/apache/qpid/client/AMQConnectionFactory.java @@ -42,7 +42,7 @@ public class AMQConnectionFactory implements ConnectionFactory, QueueConnectionF private String _virtualPath; private ConnectionURL _connectionDetails; - + private SSLConfiguration _sslConfig; public AMQConnectionFactory() { @@ -113,6 +113,22 @@ public class AMQConnectionFactory implements ConnectionFactory, QueueConnectionF } _defaultPassword = password; } + + /** + * Getter for SSLConfiguration + * @return SSLConfiguration if set, otherwise null + */ + public final SSLConfiguration getSSLConfiguration() { + return _sslConfig; + } + + /** + * Setter for SSLConfiguration + * @param sslConfig config to store + */ + public final void setSSLConfiguration(SSLConfiguration sslConfig) { + _sslConfig = sslConfig; + } /** * @return The _defaultPassword. @@ -229,7 +245,7 @@ public class AMQConnectionFactory implements ConnectionFactory, QueueConnectionF { _connectionDetails.setClientName(getUniqueClientID()); } - return new AMQConnection(_connectionDetails); + return new AMQConnection(_connectionDetails, _sslConfig); } else { @@ -260,7 +276,7 @@ public class AMQConnectionFactory implements ConnectionFactory, QueueConnectionF { _connectionDetails.setClientName(getUniqueClientID()); } - return new AMQConnection(_connectionDetails); + return new AMQConnection(_connectionDetails, _sslConfig); } else { diff --git a/java/client/src/main/java/org/apache/qpid/client/SSLConfiguration.java b/java/client/src/main/java/org/apache/qpid/client/SSLConfiguration.java new file mode 100644 index 0000000000..2280cc9870 --- /dev/null +++ b/java/client/src/main/java/org/apache/qpid/client/SSLConfiguration.java @@ -0,0 +1,61 @@ +/* + * + * 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.client; + +public class SSLConfiguration { + + private String _keystorePath; + + private String _keystorePassword; + + private String _certType = "SunX509"; + + public void setKeystorePath(String path) + { + _keystorePath = path; + } + + public String getKeystorePath() + { + return _keystorePath; + } + + public void setKeystorePassword(String password) + { + _keystorePassword = password; + } + + public String getKeystorePassword() + { + return _keystorePassword; + } + + public void setCertType(String type) + { + _certType = type; + } + + public String getCertType() + { + return _certType; + } +} diff --git a/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java b/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java index 7b789aa09d..1af7905b3b 100644 --- a/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java @@ -115,7 +115,7 @@ public class FailoverHandler implements Runnable // if _host has value then we are performing a redirect. if (_host != null) { - failoverSucceeded = _amqProtocolHandler.getConnection().attemptReconnection(_host, _port, _amqProtocolHandler.isUseSSL()); + failoverSucceeded = _amqProtocolHandler.getConnection().attemptReconnection(_host, _port); } else { diff --git a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java index 1f83ce67c3..988a12ee78 100644 --- a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java @@ -20,35 +20,44 @@ */ package org.apache.qpid.client.protocol; +import java.util.Iterator; +import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.CountDownLatch; + import org.apache.log4j.Logger; import org.apache.mina.common.IdleStatus; import org.apache.mina.common.IoHandlerAdapter; import org.apache.mina.common.IoSession; -import org.apache.mina.common.IoServiceConfig; import org.apache.mina.filter.SSLFilter; import org.apache.mina.filter.codec.ProtocolCodecFilter; import org.apache.qpid.AMQConnectionClosedException; import org.apache.qpid.AMQDisconnectedException; import org.apache.qpid.AMQException; import org.apache.qpid.AMQTimeoutException; -import org.apache.qpid.pool.ReadWriteThreadModel; -import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.client.AMQConnection; import org.apache.qpid.client.AMQSession; +import org.apache.qpid.client.SSLConfiguration; import org.apache.qpid.client.failover.FailoverHandler; import org.apache.qpid.client.failover.FailoverState; import org.apache.qpid.client.state.AMQState; import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.state.listener.SpecificMethodFrameListener; import org.apache.qpid.codec.AMQCodecFactory; -import org.apache.qpid.framing.*; +import org.apache.qpid.framing.AMQBody; +import org.apache.qpid.framing.AMQDataBlock; +import org.apache.qpid.framing.AMQFrame; +import org.apache.qpid.framing.AMQMethodBody; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.framing.ConnectionCloseBody; +import org.apache.qpid.framing.ConnectionCloseOkBody; +import org.apache.qpid.framing.ContentBody; +import org.apache.qpid.framing.ContentHeaderBody; +import org.apache.qpid.framing.HeartbeatBody; +import org.apache.qpid.pool.ReadWriteThreadModel; import org.apache.qpid.protocol.AMQConstant; +import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.protocol.AMQMethodListener; -import org.apache.qpid.ssl.BogusSSLContextFactory; - -import java.util.Iterator; -import java.util.concurrent.CopyOnWriteArraySet; -import java.util.concurrent.CountDownLatch; +import org.apache.qpid.ssl.SSLContextFactory; public class AMQProtocolHandler extends IoHandlerAdapter @@ -62,12 +71,6 @@ public class AMQProtocolHandler extends IoHandlerAdapter private AMQConnection _connection; /** - * Used only when determining whether to add the SSL filter or not. This should be made more - * generic in future since we will potentially have many transport layer options - */ - private boolean _useSSL; - - /** * Our wrapper for a protocol session that provides access to session values * in a typesafe manner. */ @@ -99,16 +102,6 @@ public class AMQProtocolHandler extends IoHandlerAdapter _connection = con; } - public boolean isUseSSL() - { - return _useSSL; - } - - public void setUseSSL(boolean useSSL) - { - _useSSL = useSSL; - } - public void sessionCreated(IoSession session) throws Exception { _logger.debug("Protocol session created for session " + System.identityHashCode(session)); @@ -125,10 +118,11 @@ public class AMQProtocolHandler extends IoHandlerAdapter session.getFilterChain().addLast("protocolFilter", pcf); } // we only add the SSL filter where we have an SSL connection - if (_useSSL) + if (_connection.getSSLConfiguration() != null) { - //FIXME: Bogus context cannot be used in production. - SSLFilter sslFilter = new SSLFilter(BogusSSLContextFactory.getInstance(false)); + SSLConfiguration sslConfig = _connection.getSSLConfiguration(); + SSLContextFactory sslFactory = new SSLContextFactory(sslConfig.getKeystorePath(), sslConfig.getKeystorePassword(), sslConfig.getCertType()); + SSLFilter sslFilter = new SSLFilter(sslFactory.buildClientContext()); sslFilter.setUseClientMode(true); session.getFilterChain().addBefore("protocolFilter", "ssl", sslFilter); } diff --git a/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java b/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java index 5e6244d7cc..9342e93ec6 100644 --- a/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java +++ b/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java @@ -81,7 +81,6 @@ public class SocketTransportConnection implements ITransportConnection scfg.setReceiveBufferSize(Integer.getInteger("amqj.receiveBufferSize", DEFAULT_BUFFER_SIZE)); _logger.info("recv-buffer-size = " + scfg.getReceiveBufferSize()); final InetSocketAddress address = new InetSocketAddress(brokerDetail.getHost(), brokerDetail.getPort()); - protocolHandler.setUseSSL(brokerDetail.useSSL()); _logger.info("Attempting connection to " + address); ConnectFuture future = ioConnector.connect(address, protocolHandler); diff --git a/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java b/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java index 293ce5e82e..91f7710025 100644 --- a/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java +++ b/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java @@ -20,6 +20,8 @@ */ package org.apache.qpid.jms; +import org.apache.qpid.client.SSLConfiguration; + public interface BrokerDetails { @@ -28,7 +30,6 @@ public interface BrokerDetails * @see ConnectionURL */ public static final String OPTIONS_RETRY = "retries"; - public static final String OPTIONS_SSL = ConnectionURL.OPTIONS_SSL; public static final String OPTIONS_CONNECT_TIMEOUT = "connecttimeout"; public static final int DEFAULT_PORT = 5672; @@ -55,10 +56,6 @@ public interface BrokerDetails void setTransport(String transport); - boolean useSSL(); - - void useSSL(boolean ssl); - String getOption(String key); void setOption(String key, String value); @@ -66,6 +63,10 @@ public interface BrokerDetails long getTimeout(); void setTimeout(long timeout); + + SSLConfiguration getSSLConfiguration(); + + void setSSLConfiguration(SSLConfiguration sslConfiguration); String toString(); |