You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2009/01/06 20:51:00 UTC
svn commit: r732082 - in /qpid/trunk/qpid/cpp/src: ./ qpid/broker/
qpid/client/ qpid/sys/ qpid/sys/cyrus/ tests/
Author: gsim
Date: Tue Jan 6 11:50:59 2009
New Revision: 732082
URL: http://svn.apache.org/viewvc?rev=732082&view=rev
Log:
* Cyrus SASL intgeration for c++ client
* SASL security layer support for c++ client and broker
Added:
qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnection.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnection.h
qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnectionFactory.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnectionFactory.h
qpid/trunk/qpid/cpp/src/qpid/client/Sasl.h
qpid/trunk/qpid/cpp/src/qpid/client/SaslFactory.cpp
qpid/trunk/qpid/cpp/src/qpid/client/SaslFactory.h
qpid/trunk/qpid/cpp/src/qpid/sys/Codec.h
qpid/trunk/qpid/cpp/src/qpid/sys/SecurityLayer.h
qpid/trunk/qpid/cpp/src/qpid/sys/cyrus/
qpid/trunk/qpid/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.cpp
qpid/trunk/qpid/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.h
Modified:
qpid/trunk/qpid/cpp/src/Makefile.am
qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Connection.h
qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionHandler.h
qpid/trunk/qpid/cpp/src/qpid/broker/SaslAuthenticator.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/SaslAuthenticator.h
qpid/trunk/qpid/cpp/src/qpid/client/ConnectionHandler.cpp
qpid/trunk/qpid/cpp/src/qpid/client/ConnectionHandler.h
qpid/trunk/qpid/cpp/src/qpid/client/ConnectionImpl.cpp
qpid/trunk/qpid/cpp/src/qpid/client/ConnectionSettings.cpp
qpid/trunk/qpid/cpp/src/qpid/client/ConnectionSettings.h
qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp
qpid/trunk/qpid/cpp/src/qpid/client/Connector.h
qpid/trunk/qpid/cpp/src/qpid/client/RdmaConnector.cpp
qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp
qpid/trunk/qpid/cpp/src/qpid/sys/ConnectionCodec.h
qpid/trunk/qpid/cpp/src/qpid/sys/TCPIOPlugin.cpp
qpid/trunk/qpid/cpp/src/tests/.valgrind.supp
qpid/trunk/qpid/cpp/src/tests/ConnectionOptions.h
Modified: qpid/trunk/qpid/cpp/src/Makefile.am
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/Makefile.am?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/Makefile.am (original)
+++ qpid/trunk/qpid/cpp/src/Makefile.am Tue Jan 6 11:50:59 2009
@@ -353,11 +353,14 @@
qpid/sys/Runnable.cpp \
qpid/sys/Shlib.cpp
-libqpidbroker_la_LIBADD = libqpidcommon.la -luuid
if HAVE_SASL
-libqpidbroker_la_LIBADD += -lsasl2
+libqpidcommon_la_SOURCES += qpid/sys/cyrus/CyrusSecurityLayer.h
+libqpidcommon_la_SOURCES += qpid/sys/cyrus/CyrusSecurityLayer.cpp
+libqpidcommon_la_LIBADD += -lsasl2
endif
+libqpidbroker_la_LIBADD = libqpidcommon.la -luuid
+
libqpidbroker_la_SOURCES = \
$(mgen_broker_cpp) \
$(posix_broker_src) \
@@ -403,6 +406,8 @@
qpid/broker/RecoveredEnqueue.cpp \
qpid/broker/RecoveredDequeue.cpp \
qpid/broker/SaslAuthenticator.cpp \
+ qpid/broker/SecureConnection.cpp \
+ qpid/broker/SecureConnectionFactory.cpp \
qpid/broker/SemanticState.h \
qpid/broker/SemanticState.cpp \
qpid/broker/SessionAdapter.cpp \
@@ -426,6 +431,7 @@
qpid/management/ManagementExchange.cpp \
qpid/sys/TCPIOPlugin.cpp
+
libqpidclient_la_LIBADD = libqpidcommon.la -luuid
libqpidclient_la_SOURCES = \
@@ -452,6 +458,7 @@
qpid/client/MessageReplayTracker.cpp \
qpid/client/QueueOptions.cpp \
qpid/client/Results.cpp \
+ qpid/client/SaslFactory.cpp \
qpid/client/SessionBase_0_10.cpp \
qpid/client/SessionBase_0_10.h \
qpid/client/SessionBase_0_10Access.h \
@@ -552,6 +559,8 @@
qpid/broker/RecoveryManager.h \
qpid/broker/RecoveryManagerImpl.h \
qpid/broker/SaslAuthenticator.h \
+ qpid/broker/SecureConnection.h \
+ qpid/broker/SecureConnectionFactory.h \
qpid/broker/SessionAdapter.h \
qpid/broker/SessionManager.h \
qpid/broker/System.h \
@@ -591,6 +600,8 @@
qpid/client/MessageListener.h \
qpid/client/MessageReplayTracker.h \
qpid/client/Results.h \
+ qpid/client/Sasl.h \
+ qpid/client/SaslFactory.h \
qpid/client/SessionBase_0_10.h \
qpid/client/Session.h \
qpid/client/SessionImpl.h \
@@ -668,6 +679,7 @@
qpid/sys/AtomicValue_gcc.h \
qpid/sys/AtomicValue_mutex.h \
qpid/sys/BlockingQueue.h \
+ qpid/sys/Codec.h \
qpid/sys/CopyOnWriteArray.h \
qpid/sys/Condition.h \
qpid/sys/ConnectionCodec.h \
@@ -692,6 +704,7 @@
qpid/sys/Runnable.h \
qpid/sys/Fork.h \
qpid/sys/ScopedIncrement.h \
+ qpid/sys/SecurityLayer.h \
qpid/sys/Semaphore.h \
qpid/sys/SystemInfo.h \
qpid/sys/Shlib.h \
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp Tue Jan 6 11:50:59 2009
@@ -27,6 +27,7 @@
#include "NullMessageStore.h"
#include "RecoveryManagerImpl.h"
#include "SaslAuthenticator.h"
+#include "SecureConnectionFactory.h"
#include "TopicExchange.h"
#include "Link.h"
@@ -135,7 +136,7 @@
acl(0),
dataDir(conf.noDataDir ? std::string() : conf.dataDir),
links(this),
- factory(new ConnectionFactory(*this)),
+ factory(new SecureConnectionFactory(*this)),
dtxManager(timer),
sessionManager(
qpid::SessionState::Configuration(
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp Tue Jan 6 11:50:59 2009
@@ -267,5 +267,10 @@
return status;
}
+void Connection::setSecureConnection(SecureConnection* s)
+{
+ adapter.setSecureConnection(s);
+}
+
}}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Connection.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.h?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Connection.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Connection.h Tue Jan 6 11:50:59 2009
@@ -57,6 +57,7 @@
namespace broker {
class LinkRegistry;
+class SecureConnection;
class Connection : public sys::ConnectionInputHandler,
public ConnectionState,
@@ -105,7 +106,7 @@
}
void sendClose();
-
+ void setSecureConnection(SecureConnection* secured);
private:
typedef boost::ptr_map<framing::ChannelId, SessionHandler> ChannelMap;
typedef std::vector<Queue::shared_ptr>::iterator queue_iterator;
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp Tue Jan 6 11:50:59 2009
@@ -22,17 +22,20 @@
#include "ConnectionHandler.h"
#include "Connection.h"
+#include "SecureConnection.h"
+#include "qpid/Url.h"
#include "qpid/framing/ClientInvoker.h"
#include "qpid/framing/ServerInvoker.h"
#include "qpid/framing/enum.h"
#include "qpid/log/Statement.h"
-#include "qpid/Url.h"
+#include "qpid/sys/SecurityLayer.h"
#include "AclModule.h"
#include "qmf/org/apache/qpid/broker/EventClientConnectFail.h"
using namespace qpid;
using namespace qpid::broker;
using namespace qpid::framing;
+using qpid::sys::SecurityLayer;
namespace _qmf = qmf::org::apache::qpid::broker;
namespace
@@ -70,11 +73,16 @@
}
}
+void ConnectionHandler::setSecureConnection(SecureConnection* secured)
+{
+ handler->secured = secured;
+}
+
ConnectionHandler::ConnectionHandler(Connection& connection, bool isClient) : handler(new Handler(connection, isClient)) {}
ConnectionHandler::Handler::Handler(Connection& c, bool isClient) :
client(c.getOutput()), server(c.getOutput()),
- connection(c), serverMode(!isClient), acl(0)
+ connection(c), serverMode(!isClient), acl(0), secured(0)
{
if (serverMode) {
@@ -160,6 +168,12 @@
for (std::vector<Url>::iterator i = urls.begin(); i < urls.end(); ++i)
array.add(boost::shared_ptr<Str16Value>(new Str16Value(i->str())));
client.openOk(array);
+
+ //install security layer if one has been negotiated:
+ if (secured) {
+ std::auto_ptr<SecurityLayer> sl = authenticator->getSecurityLayer(connection.getFrameMax());
+ if (sl.get()) secured->activateSecurityLayer(sl);
+ }
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionHandler.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionHandler.h?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionHandler.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionHandler.h Tue Jan 6 11:50:59 2009
@@ -40,6 +40,7 @@
namespace broker {
class Connection;
+class SecureConnection;
class ConnectionHandler : public framing::FrameHandler
{
@@ -52,6 +53,7 @@
bool serverMode;
std::auto_ptr<SaslAuthenticator> authenticator;
AclModule* acl;
+ SecureConnection* secured;
Handler(Connection& connection, bool isClient);
~Handler();
@@ -87,6 +89,7 @@
ConnectionHandler(Connection& connection, bool isClient);
void close(framing::connection::CloseCode code, const std::string& text);
void handle(framing::AMQFrame& frame);
+ void setSecureConnection(SecureConnection* secured);
};
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/SaslAuthenticator.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SaslAuthenticator.cpp?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SaslAuthenticator.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SaslAuthenticator.cpp Tue Jan 6 11:50:59 2009
@@ -30,9 +30,12 @@
#if HAVE_SASL
#include <sasl/sasl.h>
+#include "qpid/sys/cyrus/CyrusSecurityLayer.h"
+using qpid::sys::cyrus::CyrusSecurityLayer;
#endif
using namespace qpid::framing;
+using qpid::sys::SecurityLayer;
using boost::format;
using boost::str;
@@ -46,11 +49,12 @@
framing::AMQP_ClientProxy::Connection client;
std::string realm;
public:
- NullAuthenticator(Connection& connection);
+ NullAuthenticator(Connection& connection, bool dummy=false/*dummy arg to match CyrusAuthenticator*/);
~NullAuthenticator();
void getMechanisms(framing::Array& mechanisms);
void start(const std::string& mechanism, const std::string& response);
void step(const std::string&) {}
+ std::auto_ptr<SecurityLayer> getSecurityLayer(uint16_t maxFrameSize);
};
#if HAVE_SASL
@@ -60,11 +64,12 @@
sasl_conn_t *sasl_conn;
Connection& connection;
framing::AMQP_ClientProxy::Connection client;
+ const bool encrypt;
void processAuthenticationStep(int code, const char *challenge, unsigned int challenge_len);
public:
- CyrusAuthenticator(Connection& connection);
+ CyrusAuthenticator(Connection& connection, bool encrypt);
~CyrusAuthenticator();
void init();
void getMechanisms(framing::Array& mechanisms);
@@ -72,6 +77,7 @@
void step(const std::string& response);
void getUid(std::string& uid);
void getError(std::string& error);
+ std::auto_ptr<SecurityLayer> getSecurityLayer(uint16_t maxFrameSize);
};
bool SaslAuthenticator::available(void)
@@ -120,7 +126,7 @@
{
static bool needWarning = true;
if (c.getBroker().getOptions().auth) {
- return std::auto_ptr<SaslAuthenticator>(new CyrusAuthenticator(c));
+ return std::auto_ptr<SaslAuthenticator>(new CyrusAuthenticator(c, c.getBroker().getOptions().requireEncrypted));
} else {
QPID_LOG(warning, "SASL: No Authentication Performed");
needWarning = false;
@@ -128,7 +134,7 @@
}
}
-NullAuthenticator::NullAuthenticator(Connection& c) : connection(c), client(c.getOutput()),
+ NullAuthenticator::NullAuthenticator(Connection& c, bool /*dummy*/) : connection(c), client(c.getOutput()),
realm(c.getBroker().getOptions().realm) {}
NullAuthenticator::~NullAuthenticator() {}
@@ -158,9 +164,18 @@
}
+std::auto_ptr<SecurityLayer> NullAuthenticator::getSecurityLayer(uint16_t)
+{
+ std::auto_ptr<SecurityLayer> securityLayer;
+ return securityLayer;
+}
+
+
#if HAVE_SASL
-CyrusAuthenticator::CyrusAuthenticator(Connection& c) : sasl_conn(0), connection(c), client(c.getOutput())
+
+CyrusAuthenticator::CyrusAuthenticator(Connection& c, bool _encrypt) :
+ sasl_conn(0), connection(c), client(c.getOutput()), encrypt(_encrypt)
{
init();
}
@@ -196,6 +211,25 @@
// server error, when one is available
throw ConnectionForcedException("Unable to perform authentication");
}
+
+ sasl_security_properties_t secprops;
+
+ //TODO: should the actual SSF values be configurable here?
+ secprops.min_ssf = encrypt ? 10: 0;
+ secprops.max_ssf = 256;
+ secprops.maxbufsize = 65535;
+
+ QPID_LOG(debug, "min_ssf: " << secprops.min_ssf << ", max_ssf: " << secprops.max_ssf);
+
+ secprops.property_names = 0;
+ secprops.property_values = 0;
+ secprops.security_flags = 0; /* or SASL_SEC_NOANONYMOUS etc as appropriate */
+
+ int result = sasl_setprop(sasl_conn, SASL_SEC_PROPS, &secprops);
+ if (result != SASL_OK) {
+ throw framing::InternalErrorException(QPID_MSG("SASL error: " << result));
+ }
+
}
CyrusAuthenticator::~CyrusAuthenticator()
@@ -332,6 +366,24 @@
}
}
}
+
+std::auto_ptr<SecurityLayer> CyrusAuthenticator::getSecurityLayer(uint16_t maxFrameSize)
+{
+
+ const void* value(0);
+ int result = sasl_getprop(sasl_conn, SASL_SSF, &value);
+ if (result != SASL_OK) {
+ throw framing::InternalErrorException(QPID_MSG("SASL error: " << sasl_errdetail(sasl_conn)));
+ }
+ uint ssf = *(reinterpret_cast<const unsigned*>(value));
+ std::auto_ptr<SecurityLayer> securityLayer;
+ if (ssf) {
+ QPID_LOG(info, "Installing security layer, SSF: "<< ssf);
+ securityLayer = std::auto_ptr<SecurityLayer>(new CyrusSecurityLayer(sasl_conn, maxFrameSize));
+ }
+ return securityLayer;
+}
+
#endif
}}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/SaslAuthenticator.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SaslAuthenticator.h?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SaslAuthenticator.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SaslAuthenticator.h Tue Jan 6 11:50:59 2009
@@ -24,6 +24,7 @@
#include "qpid/framing/amqp_types.h"
#include "qpid/framing/AMQP_ClientProxy.h"
#include "qpid/Exception.h"
+#include "qpid/sys/SecurityLayer.h"
#include <memory>
namespace qpid {
@@ -40,6 +41,7 @@
virtual void step(const std::string& response) = 0;
virtual void getUid(std::string&) {}
virtual void getError(std::string&) {}
+ virtual std::auto_ptr<qpid::sys::SecurityLayer> getSecurityLayer(uint16_t maxFrameSize) = 0;
static bool available(void);
Added: qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnection.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnection.cpp?rev=732082&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnection.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnection.cpp Tue Jan 6 11:50:59 2009
@@ -0,0 +1,87 @@
+/*
+ *
+ * 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.
+ *
+ */
+#include "SecureConnection.h"
+#include "qpid/sys/SecurityLayer.h"
+#include "qpid/framing/reply_exceptions.h"
+
+namespace qpid {
+namespace broker {
+
+using qpid::sys::SecurityLayer;
+
+SecureConnection::SecureConnection() : secured(false) {}
+
+size_t SecureConnection::decode(const char* buffer, size_t size)
+{
+ if (!secured && securityLayer.get()) {
+ //security layer comes into effect on first read after its
+ //activated
+ secured = true;
+ }
+ if (secured) {
+ return securityLayer->decode(buffer, size);
+ } else {
+ return codec->decode(buffer, size);
+ }
+}
+
+size_t SecureConnection::encode(const char* buffer, size_t size)
+{
+ if (secured) {
+ return securityLayer->encode(buffer, size);
+ } else {
+ return codec->encode(buffer, size);
+ }
+}
+
+bool SecureConnection::canEncode()
+{
+ if (secured) return securityLayer->canEncode();
+ else return codec->canEncode();
+}
+
+void SecureConnection::closed()
+{
+ codec->closed();
+}
+
+bool SecureConnection::isClosed() const
+{
+ return codec->isClosed();
+}
+
+framing::ProtocolVersion SecureConnection::getVersion() const
+{
+ return codec->getVersion();
+}
+
+void SecureConnection:: setCodec(std::auto_ptr<ConnectionCodec> c)
+{
+ codec = c;
+}
+
+void SecureConnection::activateSecurityLayer(std::auto_ptr<SecurityLayer> sl)
+{
+ securityLayer = sl;
+ securityLayer->init(codec.get());
+}
+
+}} // namespace qpid::broker
Added: qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnection.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnection.h?rev=732082&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnection.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnection.h Tue Jan 6 11:50:59 2009
@@ -0,0 +1,60 @@
+#ifndef QPID_BROKER_SECURECONNECTION_H
+#define QPID_BROKER_SECURECONNECTION_H
+
+/*
+ *
+ * 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.
+ *
+ */
+
+#include "qpid/sys/ConnectionCodec.h"
+#include <memory>
+
+namespace qpid {
+
+namespace sys {
+class SecurityLayer;
+}
+
+namespace broker {
+
+/**
+ * A ConnectionCodec 'wrapper' that allows a connection to be
+ * 'secured' e.g. encrypted based on settings negotiatiated at the
+ * time of establishment.
+ */
+class SecureConnection : public qpid::sys::ConnectionCodec
+{
+ public:
+ SecureConnection();
+ size_t decode(const char* buffer, size_t size);
+ size_t encode(const char* buffer, size_t size);
+ bool canEncode();
+ void closed();
+ bool isClosed() const;
+ framing::ProtocolVersion getVersion() const;
+ void setCodec(std::auto_ptr<ConnectionCodec>);
+ void activateSecurityLayer(std::auto_ptr<qpid::sys::SecurityLayer>);
+ private:
+ std::auto_ptr<ConnectionCodec> codec;
+ std::auto_ptr<qpid::sys::SecurityLayer> securityLayer;
+ bool secured;
+};
+}} // namespace qpid::broker
+
+#endif /*!QPID_BROKER_SECURECONNECTION_H*/
Added: qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnectionFactory.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnectionFactory.cpp?rev=732082&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnectionFactory.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnectionFactory.cpp Tue Jan 6 11:50:59 2009
@@ -0,0 +1,65 @@
+/*
+ *
+ * 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.
+ *
+ */
+#include "SecureConnectionFactory.h"
+#include "qpid/framing/ProtocolVersion.h"
+#include "qpid/amqp_0_10/Connection.h"
+#include "qpid/broker/Connection.h"
+#include "qpid/broker/SecureConnection.h"
+
+namespace qpid {
+namespace broker {
+
+using framing::ProtocolVersion;
+typedef std::auto_ptr<amqp_0_10::Connection> CodecPtr;
+typedef std::auto_ptr<SecureConnection> SecureConnectionPtr;
+typedef std::auto_ptr<Connection> ConnectionPtr;
+typedef std::auto_ptr<sys::ConnectionInputHandler> InputPtr;
+
+SecureConnectionFactory::SecureConnectionFactory(Broker& b) : broker(b) {}
+
+sys::ConnectionCodec*
+SecureConnectionFactory::create(ProtocolVersion v, sys::OutputControl& out, const std::string& id) {
+ if (v == ProtocolVersion(0, 10)) {
+ SecureConnectionPtr sc(new SecureConnection());
+ CodecPtr c(new amqp_0_10::Connection(out, id, false));
+ ConnectionPtr i(new broker::Connection(c.get(), broker, id, false));
+ i->setSecureConnection(sc.get());
+ c->setInputHandler(InputPtr(i.release()));
+ sc->setCodec(std::auto_ptr<sys::ConnectionCodec>(c));
+ return sc.release();
+ }
+ return 0;
+}
+
+sys::ConnectionCodec*
+SecureConnectionFactory::create(sys::OutputControl& out, const std::string& id) {
+ // used to create connections from one broker to another
+ SecureConnectionPtr sc(new SecureConnection());
+ CodecPtr c(new amqp_0_10::Connection(out, id, true));
+ ConnectionPtr i(new broker::Connection(c.get(), broker, id, true));
+ i->setSecureConnection(sc.get());
+ c->setInputHandler(InputPtr(i.release()));
+ sc->setCodec(std::auto_ptr<sys::ConnectionCodec>(c));
+ return sc.release();
+}
+
+
+}} // namespace qpid::broker
Added: qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnectionFactory.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnectionFactory.h?rev=732082&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnectionFactory.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SecureConnectionFactory.h Tue Jan 6 11:50:59 2009
@@ -0,0 +1,48 @@
+/*
+ *
+ * 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.
+ *
+ */
+#ifndef _SecureConnectionFactory_
+#define _SecureConnectionFactory_
+
+#include "qpid/sys/ConnectionCodec.h"
+
+namespace qpid {
+namespace broker {
+class Broker;
+
+class SecureConnectionFactory : public sys::ConnectionCodec::Factory
+{
+ public:
+ SecureConnectionFactory(Broker& b);
+
+ sys::ConnectionCodec*
+ create(framing::ProtocolVersion, sys::OutputControl&, const std::string& id);
+
+ sys::ConnectionCodec*
+ create(sys::OutputControl&, const std::string& id);
+
+ private:
+ Broker& broker;
+};
+
+}}
+
+
+#endif
Modified: qpid/trunk/qpid/cpp/src/qpid/client/ConnectionHandler.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/ConnectionHandler.cpp?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/ConnectionHandler.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/ConnectionHandler.cpp Tue Jan 6 11:50:59 2009
@@ -21,16 +21,18 @@
#include "ConnectionHandler.h"
-#include "qpid/log/Statement.h"
+#include "SaslFactory.h"
#include "qpid/framing/amqp_framing.h"
#include "qpid/framing/all_method_bodies.h"
#include "qpid/framing/ClientInvoker.h"
#include "qpid/framing/reply_exceptions.h"
#include "qpid/log/Helpers.h"
+#include "qpid/log/Statement.h"
using namespace qpid::client;
using namespace qpid::framing;
using namespace qpid::framing::connection;
+using qpid::sys::SecurityLayer;
namespace {
const std::string OK("OK");
@@ -146,18 +148,50 @@
setState(FAILED);
}
-void ConnectionHandler::start(const FieldTable& /*serverProps*/, const Array& /*mechanisms*/, const Array& /*locales*/)
+namespace {
+std::string SPACE(" ");
+}
+
+void ConnectionHandler::start(const FieldTable& /*serverProps*/, const Array& mechanisms, const Array& /*locales*/)
{
checkState(NOT_STARTED, INVALID_STATE_START);
setState(NEGOTIATING);
- //TODO: verify that desired mechanism and locale are supported
- string response = ((char)0) + username + ((char)0) + password;
- proxy.startOk(properties, mechanism, response, locale);
+ sasl = SaslFactory::getInstance().create(*this);
+
+ std::string mechlist;
+ bool chosenMechanismSupported = mechanism.empty();
+ for (Array::const_iterator i = mechanisms.begin(); i != mechanisms.end(); ++i) {
+ if (!mechanism.empty() && mechanism == (*i)->get<std::string>()) {
+ chosenMechanismSupported = true;
+ mechlist = (*i)->get<std::string>() + SPACE + mechlist;
+ } else {
+ if (i != mechanisms.begin()) mechlist += SPACE;
+ mechlist += (*i)->get<std::string>();
+ }
+ }
+
+ if (!chosenMechanismSupported) {
+ fail("Selected mechanism not supported: " + mechanism);
+ }
+
+ if (sasl.get()) {
+ string response = sasl->start(mechanism.empty() ? mechlist : mechanism);
+ proxy.startOk(properties, sasl->getMechanism(), response, locale);
+ } else {
+ //TODO: verify that desired mechanism and locale are supported
+ string response = ((char)0) + username + ((char)0) + password;
+ proxy.startOk(properties, mechanism, response, locale);
+ }
}
-void ConnectionHandler::secure(const std::string& /*challenge*/)
+void ConnectionHandler::secure(const std::string& challenge)
{
- throw NotImplementedException("Challenge-response cycle not yet implemented in client");
+ if (sasl.get()) {
+ string response = sasl->step(challenge);
+ proxy.secureOk(response);
+ } else {
+ throw NotImplementedException("Challenge-response cycle not yet implemented in client");
+ }
}
void ConnectionHandler::tune(uint16_t maxChannelsProposed, uint16_t maxFrameSizeProposed,
@@ -179,6 +213,9 @@
framing::Array::ValueVector::const_iterator i;
for ( i = knownBrokers.begin(); i != knownBrokers.end(); ++i )
knownBrokersUrls.push_back(Url((*i)->get<std::string>()));
+ if (sasl.get()) {
+ securityLayer = sasl->getSecurityLayer(maxFrameSize);
+ }
setState(OPEN);
QPID_LOG(debug, "Known-brokers for connection: " << log::formatList(knownBrokersUrls));
}
@@ -224,3 +261,8 @@
}
bool ConnectionHandler::isClosing() const { return getState() == CLOSING; }
+
+std::auto_ptr<qpid::sys::SecurityLayer> ConnectionHandler::getSecurityLayer()
+{
+ return securityLayer;
+}
Modified: qpid/trunk/qpid/cpp/src/qpid/client/ConnectionHandler.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/ConnectionHandler.h?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/ConnectionHandler.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/ConnectionHandler.h Tue Jan 6 11:50:59 2009
@@ -23,6 +23,7 @@
#include "ChainableFrameHandler.h"
#include "ConnectionSettings.h"
+#include "Sasl.h"
#include "StateManager.h"
#include "qpid/framing/AMQMethodBody.h"
#include "qpid/framing/AMQP_HighestVersion.h"
@@ -33,7 +34,9 @@
#include "qpid/framing/FieldTable.h"
#include "qpid/framing/FrameHandler.h"
#include "qpid/framing/InputHandler.h"
+#include "qpid/sys/SecurityLayer.h"
#include "qpid/Url.h"
+#include <memory>
namespace qpid {
namespace client {
@@ -64,6 +67,8 @@
framing::ProtocolVersion version;
framing::Array capabilities;
framing::FieldTable properties;
+ std::auto_ptr<Sasl> sasl;
+ std::auto_ptr<qpid::sys::SecurityLayer> securityLayer;
void checkState(STATES s, const std::string& msg);
@@ -103,6 +108,8 @@
bool isClosed() const;
bool isClosing() const;
+ std::auto_ptr<qpid::sys::SecurityLayer> getSecurityLayer();
+
CloseListener onClose;
ErrorListener onError;
Modified: qpid/trunk/qpid/cpp/src/qpid/client/ConnectionImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/ConnectionImpl.cpp?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/ConnectionImpl.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/ConnectionImpl.cpp Tue Jan 6 11:50:59 2009
@@ -110,6 +110,14 @@
connector->connect(host, port);
connector->init();
handler.waitForOpen();
+ //enable security layer if one has been negotiated:
+ std::auto_ptr<SecurityLayer> securityLayer = handler.getSecurityLayer();
+ if (securityLayer.get()) {
+ QPID_LOG(debug, "Activating security layer");
+ connector->activateSecurityLayer(securityLayer);
+ } else {
+ QPID_LOG(debug, "No security layer in place");
+ }
failover.reset(new FailoverListener(shared_from_this(), handler.knownBrokersUrls));
}
Modified: qpid/trunk/qpid/cpp/src/qpid/client/ConnectionSettings.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/ConnectionSettings.cpp?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/ConnectionSettings.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/ConnectionSettings.cpp Tue Jan 6 11:50:59 2009
@@ -22,6 +22,7 @@
#include "qpid/log/Logger.h"
#include "qpid/sys/Socket.h"
+#include "qpid/Version.h"
namespace qpid {
namespace client {
@@ -30,15 +31,15 @@
protocol("tcp"),
host("localhost"),
port(TcpAddress::DEFAULT_PORT),
- username("guest"),
- password("guest"),
- mechanism("PLAIN"),
locale("en_US"),
heartbeat(0),
maxChannels(32767),
maxFrameSize(65535),
bounds(2),
- tcpNoDelay(false)
+ tcpNoDelay(false),
+ service(qpid::saslName),
+ minSsf(0),
+ maxSsf(256)
{}
ConnectionSettings::~ConnectionSettings() {}
Modified: qpid/trunk/qpid/cpp/src/qpid/client/ConnectionSettings.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/ConnectionSettings.h?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/ConnectionSettings.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/ConnectionSettings.h Tue Jan 6 11:50:59 2009
@@ -71,7 +71,8 @@
std::string virtualhost;
/**
- * The username to use when authenticating the connection.
+ * The username to use when authenticating the connection. If not
+ * specified the current users login is used if available.
*/
std::string username;
/**
@@ -111,6 +112,20 @@
* If true, TCP_NODELAY will be set for the connection.
*/
bool tcpNoDelay;
+ /**
+ * SASL service name
+ */
+ std::string service;
+ /**
+ * Minimum acceptable strength of any SASL negotiated security
+ * layer. 0 means no security layer required.
+ */
+ uint minSsf;
+ /**
+ * Maximum acceptable strength of any SASL negotiated security
+ * layer. 0 means no security layer allowed.
+ */
+ uint maxSsf;
};
}} // namespace qpid::client
Modified: qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp Tue Jan 6 11:50:59 2009
@@ -24,15 +24,18 @@
#include "ConnectionImpl.h"
#include "ConnectionSettings.h"
#include "qpid/log/Statement.h"
+#include "qpid/sys/Codec.h"
#include "qpid/sys/Time.h"
#include "qpid/framing/AMQFrame.h"
#include "qpid/sys/AsynchIO.h"
#include "qpid/sys/Dispatcher.h"
#include "qpid/sys/Poller.h"
+#include "qpid/sys/SecurityLayer.h"
#include "qpid/Msg.h"
#include <iostream>
#include <map>
+#include <deque>
#include <boost/bind.hpp>
#include <boost/format.hpp>
#include <boost/weak_ptr.hpp>
@@ -74,39 +77,19 @@
theProtocolRegistry()[proto] = connectorFactory;
}
-class TCPConnector : public Connector, private sys::Runnable
+class TCPConnector : public Connector, public sys::Codec, private sys::Runnable
{
+ typedef std::deque<framing::AMQFrame> Frames;
struct Buff;
- /** Batch up frames for writing to aio. */
- class Writer : public framing::FrameHandler {
- typedef sys::AsynchIOBufferBase BufferBase;
- typedef std::vector<framing::AMQFrame> Frames;
-
- const uint16_t maxFrameSize;
- sys::Mutex lock;
- sys::AsynchIO* aio;
- BufferBase* buffer;
- Frames frames;
- size_t lastEof; // Position after last EOF in frames
- framing::Buffer encode;
- size_t framesEncoded;
- std::string identifier;
- Bounds* bounds;
-
- void writeOne();
- void newBuffer();
+ const uint16_t maxFrameSize;
- public:
-
- Writer(uint16_t maxFrameSize, Bounds*);
- ~Writer();
- void init(std::string id, sys::AsynchIO*);
- void handle(framing::AMQFrame&);
- void write(sys::AsynchIO&);
- };
+ sys::Mutex lock;
+ Frames frames; // Outgoing frame queue
+ size_t lastEof; // Position after last EOF in frames
+ uint64_t currentSize;
+ Bounds* bounds;
- const uint16_t maxFrameSize;
framing::ProtocolVersion version;
bool initiated;
@@ -119,14 +102,14 @@
framing::InitiationHandler* initialiser;
framing::OutputHandler* output;
- Writer writer;
-
sys::Thread receiver;
sys::Socket socket;
sys::AsynchIO* aio;
+ std::string identifier;
boost::shared_ptr<sys::Poller> poller;
+ std::auto_ptr<qpid::sys::SecurityLayer> securityLayer;
~TCPConnector();
@@ -139,8 +122,6 @@
void writeDataBlock(const framing::AMQDataBlock& data);
void eof(qpid::sys::AsynchIO&);
- std::string identifier;
-
boost::weak_ptr<ConnectionImpl> impl;
void connect(const std::string& host, int port);
@@ -153,6 +134,12 @@
sys::ShutdownHandler* getShutdownHandler() const;
framing::OutputHandler* getOutputHandler();
const std::string& getIdentifier() const;
+ void activateSecurityLayer(std::auto_ptr<qpid::sys::SecurityLayer>);
+
+ size_t decode(const char* buffer, size_t size);
+ size_t encode(const char* buffer, size_t size);
+ bool canEncode();
+
public:
TCPConnector(framing::ProtocolVersion pVersion,
@@ -177,12 +164,14 @@
const ConnectionSettings& settings,
ConnectionImpl* cimpl)
: maxFrameSize(settings.maxFrameSize),
+ lastEof(0),
+ currentSize(0),
+ bounds(cimpl),
version(ver),
initiated(false),
closed(true),
joined(true),
shutdownHandler(0),
- writer(maxFrameSize, cimpl),
aio(0),
impl(cimpl->shared_from_this())
{
@@ -214,7 +203,6 @@
0, // closed
0, // nobuffs
boost::bind(&TCPConnector::writebuff, this, _1));
- writer.init(identifier, aio);
}
void TCPConnector::init(){
@@ -266,7 +254,21 @@
}
void TCPConnector::send(AMQFrame& frame) {
- writer.handle(frame);
+ bool notifyWrite = false;
+ {
+ Mutex::ScopedLock l(lock);
+ frames.push_back(frame);
+ //only ask to write if this is the end of a frameset or if we
+ //already have a buffers worth of data
+ currentSize += frame.encodedSize();
+ if (frame.getEof()) {
+ lastEof = frames.size();
+ notifyWrite = true;
+ } else {
+ notifyWrite = (currentSize >= maxFrameSize);
+ }
+ }
+ if (notifyWrite) aio->notifyPendingWrite();
}
void TCPConnector::handleClosed() {
@@ -279,70 +281,70 @@
~Buff() { delete [] bytes;}
};
-TCPConnector::Writer::Writer(uint16_t s, Bounds* b) : maxFrameSize(s), aio(0), buffer(0), lastEof(0), bounds(b)
+void TCPConnector::writebuff(AsynchIO& /*aio*/)
{
-}
-
-TCPConnector::Writer::~Writer() { delete buffer; }
+ Codec* codec = securityLayer.get() ? (Codec*) securityLayer.get() : (Codec*) this;
+ if (codec->canEncode()) {
+ std::auto_ptr<AsynchIO::BufferBase> buffer = std::auto_ptr<AsynchIO::BufferBase>(aio->getQueuedBuffer());
+ if (!buffer.get()) buffer = std::auto_ptr<AsynchIO::BufferBase>(new Buff(maxFrameSize));
+
+ size_t encoded = codec->encode(buffer->bytes, buffer->byteCount);
-void TCPConnector::Writer::init(std::string id, sys::AsynchIO* a) {
- Mutex::ScopedLock l(lock);
- identifier = id;
- aio = a;
- newBuffer();
+ buffer->dataStart = 0;
+ buffer->dataCount = encoded;
+ aio->queueWrite(buffer.release());
+ }
}
-void TCPConnector::Writer::handle(framing::AMQFrame& frame) {
+
+// Called in IO thread.
+bool TCPConnector::canEncode()
+{
Mutex::ScopedLock l(lock);
- frames.push_back(frame);
- //only try to write if this is the end of a frameset or if we
- //already have a buffers worth of data
- if (frame.getEof() || (bounds && bounds->getCurrentSize() >= maxFrameSize)) {
- lastEof = frames.size();
- aio->notifyPendingWrite();
- }
- QPID_LOG(trace, "SENT " << identifier << ": " << frame);
-}
-
-void TCPConnector::Writer::writeOne() {
- assert(buffer);
- framesEncoded = 0;
-
- buffer->dataStart = 0;
- buffer->dataCount = encode.getPosition();
- aio->queueWrite(buffer);
- newBuffer();
-}
-
-void TCPConnector::Writer::newBuffer() {
- buffer = aio->getQueuedBuffer();
- if (!buffer) buffer = new Buff(maxFrameSize);
- encode = framing::Buffer(buffer->bytes, buffer->byteCount);
- framesEncoded = 0;
+ //have at least one full frameset or a whole buffers worth of data
+ return lastEof || currentSize >= maxFrameSize;
}
// Called in IO thread.
-void TCPConnector::Writer::write(sys::AsynchIO&) {
- Mutex::ScopedLock l(lock);
- assert(buffer);
+size_t TCPConnector::encode(const char* buffer, size_t size)
+{
+ framing::Buffer out(const_cast<char*>(buffer), size);
size_t bytesWritten(0);
- for (size_t i = 0; i < lastEof; ++i) {
- AMQFrame& frame = frames[i];
- uint32_t size = frame.encodedSize();
- if (size > encode.available()) writeOne();
- assert(size <= encode.available());
- frame.encode(encode);
- ++framesEncoded;
- bytesWritten += size;
+ {
+ Mutex::ScopedLock l(lock);
+ while (!frames.empty() && out.available() >= frames.front().encodedSize() ) {
+ frames.front().encode(out);
+ QPID_LOG(trace, "SENT " << identifier << ": " << frames.front());
+ frames.pop_front();
+ if (lastEof) --lastEof;
+ }
+ bytesWritten = size - out.available();
+ currentSize -= bytesWritten;
}
- frames.erase(frames.begin(), frames.begin()+lastEof);
- lastEof = 0;
if (bounds) bounds->reduce(bytesWritten);
- if (encode.getPosition() > 0) writeOne();
+ return bytesWritten;
}
-bool TCPConnector::readbuff(AsynchIO& aio, AsynchIO::BufferBase* buff) {
- framing::Buffer in(buff->bytes+buff->dataStart, buff->dataCount);
+bool TCPConnector::readbuff(AsynchIO& aio, AsynchIO::BufferBase* buff)
+{
+ Codec* codec = securityLayer.get() ? (Codec*) securityLayer.get() : (Codec*) this;
+ int32_t decoded = codec->decode(buff->bytes+buff->dataStart, buff->dataCount);
+ // TODO: unreading needs to go away, and when we can cope
+ // with multiple sub-buffers in the general buffer scheme, it will
+ if (decoded < buff->dataCount) {
+ // Adjust buffer for used bytes and then "unread them"
+ buff->dataStart += decoded;
+ buff->dataCount -= decoded;
+ aio.unread(buff);
+ } else {
+ // Give whole buffer back to aio subsystem
+ aio.queueReadBuffer(buff);
+ }
+ return true;
+}
+size_t TCPConnector::decode(const char* buffer, size_t size)
+{
+ framing::Buffer in(const_cast<char*>(buffer), size);
if (!initiated) {
framing::ProtocolInitiation protocolInit;
if (protocolInit.decode(in)) {
@@ -356,22 +358,7 @@
QPID_LOG(trace, "RECV " << identifier << ": " << frame);
input->received(frame);
}
- // TODO: unreading needs to go away, and when we can cope
- // with multiple sub-buffers in the general buffer scheme, it will
- if (in.available() != 0) {
- // Adjust buffer for used bytes and then "unread them"
- buff->dataStart += buff->dataCount-in.available();
- buff->dataCount = in.available();
- aio.unread(buff);
- } else {
- // Give whole buffer back to aio subsystem
- aio.queueReadBuffer(buff);
- }
- return true;
-}
-
-void TCPConnector::writebuff(AsynchIO& aio_) {
- writer.write(aio_);
+ return size - in.available();
}
void TCPConnector::writeDataBlock(const AMQDataBlock& data) {
@@ -388,7 +375,7 @@
// TODO: astitcher 20070908 This version of the code can never time out, so the idle processing
// will never be called
-void TCPConnector::run(){
+void TCPConnector::run() {
// Keep the connection impl in memory until run() completes.
boost::shared_ptr<ConnectionImpl> protect = impl.lock();
assert(protect);
@@ -409,5 +396,11 @@
}
}
+void TCPConnector::activateSecurityLayer(std::auto_ptr<qpid::sys::SecurityLayer> sl)
+{
+ securityLayer = sl;
+ securityLayer->init(this);
+}
+
}} // namespace qpid::client
Modified: qpid/trunk/qpid/cpp/src/qpid/client/Connector.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/Connector.h?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/Connector.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/Connector.h Tue Jan 6 11:50:59 2009
@@ -40,6 +40,11 @@
#include <boost/shared_ptr.hpp>
namespace qpid {
+
+namespace sys {
+class SecurityLayer;
+}
+
namespace client {
struct ConnectionSettings;
@@ -65,6 +70,9 @@
virtual sys::ShutdownHandler* getShutdownHandler() const = 0;
virtual framing::OutputHandler* getOutputHandler() = 0;
virtual const std::string& getIdentifier() const = 0;
+
+ virtual void activateSecurityLayer(std::auto_ptr<qpid::sys::SecurityLayer>) {}
+
};
}}
Modified: qpid/trunk/qpid/cpp/src/qpid/client/RdmaConnector.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/RdmaConnector.cpp?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/RdmaConnector.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/RdmaConnector.cpp Tue Jan 6 11:50:59 2009
@@ -29,6 +29,7 @@
#include "qpid/sys/rdma/RdmaIO.h"
#include "qpid/sys/Dispatcher.h"
#include "qpid/sys/Poller.h"
+#include "qpid/sys/SecurityLayer.h"
#include "qpid/Msg.h"
#include <iostream>
@@ -47,39 +48,21 @@
using boost::format;
using boost::str;
-class RdmaConnector : public Connector, private sys::Runnable
+ class RdmaConnector : public Connector, public sys::Codec, private sys::Runnable
{
struct Buff;
- /** Batch up frames for writing to aio. */
- class Writer : public framing::FrameHandler {
- typedef Rdma::Buffer BufferBase;
- typedef std::deque<framing::AMQFrame> Frames;
-
- const uint16_t maxFrameSize;
- sys::Mutex lock;
- Rdma::AsynchIO* aio;
- BufferBase* buffer;
- Frames frames;
- size_t lastEof; // Position after last EOF in frames
- framing::Buffer encode;
- size_t framesEncoded;
- std::string identifier;
- Bounds* bounds;
-
- void writeOne();
- void newBuffer();
+ typedef Rdma::Buffer BufferBase;
+ typedef std::deque<framing::AMQFrame> Frames;
- public:
-
- Writer(uint16_t maxFrameSize, Bounds*);
- ~Writer();
- void init(std::string id, Rdma::AsynchIO*);
- void handle(framing::AMQFrame&);
- void write(Rdma::AsynchIO&);
- };
-
const uint16_t maxFrameSize;
+ sys::Mutex lock;
+ Frames frames;
+ size_t lastEof; // Position after last EOF in frames
+ uint64_t currentSize;
+ Bounds* bounds;
+
+
framing::ProtocolVersion version;
bool initiated;
@@ -92,12 +75,11 @@
framing::InitiationHandler* initialiser;
framing::OutputHandler* output;
- Writer writer;
-
sys::Thread receiver;
Rdma::AsynchIO* aio;
sys::Poller::shared_ptr poller;
+ std::auto_ptr<qpid::sys::SecurityLayer> securityLayer;
~RdmaConnector();
@@ -129,6 +111,11 @@
sys::ShutdownHandler* getShutdownHandler() const;
framing::OutputHandler* getOutputHandler();
const std::string& getIdentifier() const;
+ void activateSecurityLayer(std::auto_ptr<qpid::sys::SecurityLayer>);
+
+ size_t decode(const char* buffer, size_t size);
+ size_t encode(const char* buffer, size_t size);
+ bool canEncode();
public:
RdmaConnector(framing::ProtocolVersion pVersion,
@@ -155,12 +142,14 @@
const ConnectionSettings& settings,
ConnectionImpl* cimpl)
: maxFrameSize(settings.maxFrameSize),
+ lastEof(0),
+ currentSize(0),
+ bounds(cimpl),
version(ver),
initiated(false),
polling(false),
joined(true),
shutdownHandler(0),
- writer(maxFrameSize, cimpl),
aio(0),
impl(cimpl)
{
@@ -216,7 +205,6 @@
aio->start(poller);
identifier = str(format("[%1% %2%]") % ci->getLocalName() % ci->getPeerName());
- writer.init(identifier, aio);
ProtocolInitiation init(version);
writeDataBlock(init);
}
@@ -279,7 +267,21 @@
}
void RdmaConnector::send(AMQFrame& frame) {
- writer.handle(frame);
+ bool notifyWrite = false;
+ {
+ Mutex::ScopedLock l(lock);
+ frames.push_back(frame);
+ //only ask to write if this is the end of a frameset or if we
+ //already have a buffers worth of data
+ currentSize += frame.encodedSize();
+ if (frame.getEof()) {
+ lastEof = frames.size();
+ notifyWrite = true;
+ } else {
+ notifyWrite = (currentSize >= maxFrameSize);
+ }
+ }
+ if (notifyWrite) aio->notifyPendingWrite();
}
void RdmaConnector::handleClosed() {
@@ -287,88 +289,54 @@
shutdownHandler->shutdown();
}
-RdmaConnector::Writer::Writer(uint16_t s, Bounds* b) :
- maxFrameSize(s),
- aio(0),
- buffer(0),
- lastEof(0),
- bounds(b)
-{
-}
-
-RdmaConnector::Writer::~Writer() {
- if (aio)
- aio->returnBuffer(buffer);
+// Called in IO thread. (write idle routine)
+// This is NOT only called in response to previously calling notifyPendingWrite
+void RdmaConnector::writebuff(Rdma::AsynchIO&) {
+ Codec* codec = securityLayer.get() ? (Codec*) securityLayer.get() : (Codec*) this;
+ if (codec->canEncode()) {
+ std::auto_ptr<BufferBase> buffer = std::auto_ptr<BufferBase>(aio->getBuffer());
+ size_t encoded = codec->encode(buffer->bytes, buffer->byteCount);
+
+ buffer->dataStart = 0;
+ buffer->dataCount = encoded;
+ aio->queueWrite(buffer.release());
+ }
}
-void RdmaConnector::Writer::init(std::string id, Rdma::AsynchIO* a) {
- Mutex::ScopedLock l(lock);
- identifier = id;
- aio = a;
- assert(aio->bufferAvailable());
- newBuffer();
-}
-void RdmaConnector::Writer::handle(framing::AMQFrame& frame) {
+bool RdmaConnector::canEncode()
+{
Mutex::ScopedLock l(lock);
- frames.push_back(frame);
- // Don't bother to send anything unless we're at the end of a frameset (assembly in 0-10 terminology)
- if (frame.getEof()) {
- lastEof = frames.size();
- QPID_LOG(debug, "Requesting write: lastEof=" << lastEof);
- aio->notifyPendingWrite();
- }
- QPID_LOG(trace, "SENT " << identifier << ": " << frame);
-}
-
-void RdmaConnector::Writer::writeOne() {
- assert(buffer);
- QPID_LOG(trace, "Write buffer " << encode.getPosition()
- << " bytes " << framesEncoded << " frames ");
- framesEncoded = 0;
-
- buffer->dataStart = 0;
- buffer->dataCount = encode.getPosition();
- aio->queueWrite(buffer);
- newBuffer();
-}
-
-void RdmaConnector::Writer::newBuffer() {
- buffer = aio->getBuffer();
- encode = framing::Buffer(buffer->bytes, buffer->byteCount);
- framesEncoded = 0;
+ //have at least one full frameset or a whole buffers worth of data
+ return aio->writable() && aio->bufferAvailable() && (lastEof || currentSize >= maxFrameSize);
}
-// Called in IO thread. (write idle routine)
-// This is NOT only called in response to previously calling notifyPendingWrite
-void RdmaConnector::Writer::write(Rdma::AsynchIO&) {
- Mutex::ScopedLock l(lock);
- assert(buffer);
- // If nothing to do return immediately
- if (lastEof==0)
- return;
- size_t bytesWritten = 0;
- while (aio->writable() && aio->bufferAvailable() && !frames.empty()) {
- const AMQFrame* frame = &frames.front();
- uint32_t size = frame->encodedSize();
- while (size <= encode.available()) {
- frame->encode(encode);
+size_t RdmaConnector::encode(const char* buffer, size_t size)
+{
+ framing::Buffer out(const_cast<char*>(buffer), size);
+ size_t bytesWritten(0);
+ {
+ Mutex::ScopedLock l(lock);
+ while (!frames.empty() && out.available() >= frames.front().encodedSize() ) {
+ frames.front().encode(out);
+ QPID_LOG(trace, "SENT " << identifier << ": " << frames.front());
frames.pop_front();
- ++framesEncoded;
- bytesWritten += size;
- if (frames.empty())
- break;
- frame = &frames.front();
- size = frame->encodedSize();
+ if (lastEof) --lastEof;
}
- lastEof -= framesEncoded;
- writeOne();
+ bytesWritten = size - out.available();
+ currentSize -= bytesWritten;
}
if (bounds) bounds->reduce(bytesWritten);
+ return bytesWritten;
}
void RdmaConnector::readbuff(Rdma::AsynchIO&, Rdma::Buffer* buff) {
- framing::Buffer in(buff->bytes+buff->dataStart, buff->dataCount);
+ Codec* codec = securityLayer.get() ? (Codec*) securityLayer.get() : (Codec*) this;
+ codec->decode(buff->bytes+buff->dataStart, buff->dataCount);
+}
+size_t RdmaConnector::decode(const char* buffer, size_t size)
+{
+ framing::Buffer in(const_cast<char*>(buffer), size);
if (!initiated) {
framing::ProtocolInitiation protocolInit;
if (protocolInit.decode(in)) {
@@ -382,10 +350,7 @@
QPID_LOG(trace, "RECV " << identifier << ": " << frame);
input->received(frame);
}
-}
-
-void RdmaConnector::writebuff(Rdma::AsynchIO& aio_) {
- writer.write(aio_);
+ return size - in.available();
}
void RdmaConnector::writeDataBlock(const AMQDataBlock& data) {
@@ -424,5 +389,10 @@
}
}
+void RdmaConnector::activateSecurityLayer(std::auto_ptr<qpid::sys::SecurityLayer> sl)
+{
+ securityLayer = sl;
+ securityLayer->init(this);
+}
}} // namespace qpid::client
Added: qpid/trunk/qpid/cpp/src/qpid/client/Sasl.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/Sasl.h?rev=732082&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/Sasl.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/client/Sasl.h Tue Jan 6 11:50:59 2009
@@ -0,0 +1,52 @@
+#ifndef QPID_CLIENT_SASL_H
+#define QPID_CLIENT_SASL_H
+
+/*
+ *
+ * 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.
+ *
+ */
+
+#include <memory>
+#include <string>
+
+namespace qpid {
+
+namespace sys {
+class SecurityLayer;
+}
+
+namespace client {
+
+class ConnectionSettings;
+
+/**
+ * Interface to SASL support
+ */
+class Sasl
+{
+ public:
+ virtual std::string start(const std::string& mechanisms) = 0;
+ virtual std::string step(const std::string& challenge) = 0;
+ virtual std::string getMechanism() = 0;
+ virtual std::auto_ptr<qpid::sys::SecurityLayer> getSecurityLayer(uint16_t maxFrameSize) = 0;
+ virtual ~Sasl() {}
+};
+}} // namespace qpid::client
+
+#endif /*!QPID_CLIENT_SASL_H*/
Added: qpid/trunk/qpid/cpp/src/qpid/client/SaslFactory.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/SaslFactory.cpp?rev=732082&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/SaslFactory.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/client/SaslFactory.cpp Tue Jan 6 11:50:59 2009
@@ -0,0 +1,345 @@
+/*
+ *
+ * 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.
+ *
+ */
+#include "SaslFactory.h"
+#include "ConnectionSettings.h"
+
+#ifdef HAVE_CONFIG_H
+# include "config.h"
+#endif
+
+#ifndef HAVE_SASL
+
+namespace qpid {
+namespace client {
+
+//Null implementation
+
+SaslFactory::SaslFactory() {}
+
+SaslFactory::~SaslFactory() {}
+
+SaslFactory& SaslFactory::getInstance()
+{
+ qpid::sys::Mutex::ScopedLock l(lock);
+ if (!instance.get()) {
+ instance = std::auto_ptr<SaslFactory>(new SaslFactory());
+ }
+ return *instance;
+}
+
+std::auto_ptr<Sasl> SaslFactory::create(const ConnectionSettings&)
+{
+ return std::auto_ptr<Sasl>();
+}
+
+qpid::sys::Mutex SaslFactory::lock;
+std::auto_ptr<SaslFactory> SaslFactory::instance;
+
+}} // namespace qpid::client
+
+#else
+
+#include "qpid/Exception.h"
+#include "qpid/framing/reply_exceptions.h"
+#include "qpid/sys/SecurityLayer.h"
+#include "qpid/sys/cyrus/CyrusSecurityLayer.h"
+#include "qpid/log/Statement.h"
+#include <sasl/sasl.h>
+#include <strings.h>
+
+namespace qpid {
+namespace client {
+
+using qpid::sys::SecurityLayer;
+using qpid::sys::cyrus::CyrusSecurityLayer;
+using qpid::framing::InternalErrorException;
+
+const size_t MAX_LOGIN_LENGTH = 50;
+
+class CyrusSasl : public Sasl
+{
+ public:
+ CyrusSasl(const ConnectionSettings&);
+ ~CyrusSasl();
+ std::string start(const std::string& mechanisms);
+ std::string step(const std::string& challenge);
+ std::string getMechanism();
+ std::auto_ptr<SecurityLayer> getSecurityLayer(uint16_t maxFrameSize);
+ private:
+ sasl_conn_t* conn;
+ sasl_callback_t callbacks[5];//realm, user, authname, password, end-of-list
+ ConnectionSettings settings;
+ std::string input;
+ std::string mechanism;
+ char login[MAX_LOGIN_LENGTH];
+
+ void interact(sasl_interact_t* client_interact);
+};
+
+//sasl callback functions
+int getLogin(void *context, int id, const char **result, unsigned *len);
+int getUserFromSettings(void *context, int id, const char **result, unsigned *len);
+int getPasswordFromSettings(sasl_conn_t *conn, void *context, int id, sasl_secret_t **psecret);
+typedef int CallbackProc();
+
+qpid::sys::Mutex SaslFactory::lock;
+std::auto_ptr<SaslFactory> SaslFactory::instance;
+
+SaslFactory::SaslFactory()
+{
+ sasl_callback_t* callbacks = 0;
+ int result = sasl_client_init(callbacks);
+ if (result != SASL_OK) {
+ throw InternalErrorException(QPID_MSG("Sasl error: " << sasl_errstring(result, 0, 0)));
+ }
+}
+
+SaslFactory::~SaslFactory()
+{
+ sasl_done();
+}
+
+SaslFactory& SaslFactory::getInstance()
+{
+ qpid::sys::Mutex::ScopedLock l(lock);
+ if (!instance.get()) {
+ instance = std::auto_ptr<SaslFactory>(new SaslFactory());
+ }
+ return *instance;
+}
+
+std::auto_ptr<Sasl> SaslFactory::create(const ConnectionSettings& settings)
+{
+ std::auto_ptr<Sasl> sasl(new CyrusSasl(settings));
+ return sasl;
+}
+
+CyrusSasl::CyrusSasl(const ConnectionSettings& s) : conn(0), settings(s)
+{
+ size_t i = 0;
+
+ callbacks[i].id = SASL_CB_GETREALM;
+ callbacks[i].proc = 0;
+ callbacks[i++].context = 0;
+
+ if (settings.username.empty()) {
+ callbacks[i].id = SASL_CB_USER;
+ callbacks[i].proc = (CallbackProc*) &getLogin;
+ callbacks[i++].context = &login;
+
+ callbacks[i].id = SASL_CB_AUTHNAME;
+ callbacks[i].proc = (CallbackProc*) &getLogin;
+ callbacks[i++].context = &login;
+ } else {
+ callbacks[i].id = SASL_CB_USER;
+ callbacks[i].proc = (CallbackProc*) &getUserFromSettings;
+ callbacks[i++].context = &settings;
+
+ callbacks[i].id = SASL_CB_AUTHNAME;
+ callbacks[i].proc = (CallbackProc*) &getUserFromSettings;
+ callbacks[i++].context = &settings;
+ }
+
+ callbacks[i].id = SASL_CB_PASS;
+ callbacks[i].proc = (CallbackProc*) &getPasswordFromSettings;
+ callbacks[i++].context = &settings;
+
+ callbacks[i].id = SASL_CB_LIST_END;
+ callbacks[i].proc = 0;
+ callbacks[i++].context = 0;
+}
+
+CyrusSasl::~CyrusSasl()
+{
+ if (conn) {
+ sasl_dispose(&conn);
+ }
+}
+
+namespace {
+ const std::string SSL("ssl");
+}
+
+std::string CyrusSasl::start(const std::string& mechanisms)
+{
+ QPID_LOG(debug, "CyrusSasl::start(" << mechanisms << ")");
+ int result = sasl_client_new(settings.service.c_str(),
+ settings.host.c_str(),
+ 0, 0, /* Local and remote IP address strings */
+ callbacks,
+ 0, /* security flags */
+ &conn);
+
+ if (result != SASL_OK) throw InternalErrorException(QPID_MSG("Sasl error: " << sasl_errdetail(conn)));
+
+ sasl_security_properties_t secprops;
+
+ secprops.min_ssf = settings.minSsf;
+ secprops.max_ssf = settings.maxSsf;
+ secprops.maxbufsize = 65535;
+
+ QPID_LOG(debug, "min_ssf: " << secprops.min_ssf << ", max_ssf: " << secprops.max_ssf);
+
+ secprops.property_names = 0;
+ secprops.property_values = 0;
+ secprops.security_flags = 0;//TODO: provide means for application to configure these
+
+ result = sasl_setprop(conn, SASL_SEC_PROPS, &secprops);
+ if (result != SASL_OK) {
+ throw framing::InternalErrorException(QPID_MSG("SASL error: " << sasl_errdetail(conn)));
+ }
+
+
+ sasl_interact_t* client_interact = 0;
+ const char *out = 0;
+ unsigned outlen = 0;
+ const char *chosenMechanism = 0;
+
+ do {
+ result = sasl_client_start(conn,
+ mechanisms.c_str(),
+ &client_interact,
+ &out,
+ &outlen,
+ &chosenMechanism);
+
+ if (result == SASL_INTERACT) {
+ interact(client_interact);
+ }
+ } while (result == SASL_INTERACT);
+
+ if (result != SASL_CONTINUE && result != SASL_OK) {
+ throw InternalErrorException(QPID_MSG("Sasl error: " << sasl_errdetail(conn)));
+ }
+
+ mechanism = std::string(chosenMechanism);
+ QPID_LOG(debug, "CyrusSasl::start(" << mechanisms << "): selected "
+ << mechanism << " response: '" << std::string(out, outlen) << "'");
+ return std::string(out, outlen);
+}
+
+std::string CyrusSasl::step(const std::string& challenge)
+{
+ sasl_interact_t* client_interact = 0;
+ const char *out = 0;
+ unsigned outlen = 0;
+ int result = 0;
+ do {
+ result = sasl_client_step(conn, /* our context */
+ challenge.data(), /* the data from the server */
+ challenge.size(), /* it's length */
+ &client_interact, /* this should be
+ unallocated and NULL */
+ &out, /* filled in on success */
+ &outlen); /* filled in on success */
+
+ if (result == SASL_INTERACT) {
+ interact(client_interact);
+ }
+ } while (result == SASL_INTERACT);
+
+ std::string response;
+ if (result == SASL_CONTINUE || result == SASL_OK) response = std::string(out, outlen);
+ else if (result != SASL_OK) {
+ throw InternalErrorException(QPID_MSG("Sasl error: " << sasl_errdetail(conn)));
+ }
+ QPID_LOG(debug, "CyrusSasl::step(" << challenge << "): " << response);
+ return response;
+}
+
+std::string CyrusSasl::getMechanism()
+{
+ return mechanism;
+}
+
+void CyrusSasl::interact(sasl_interact_t* client_interact)
+{
+ std::cout << "[" << client_interact->id << "] " << client_interact->challenge << " " << client_interact->prompt;
+ if (client_interact->defresult) std::cout << " (" << client_interact->defresult << ")";
+ std::cout << std::endl;
+ if (std::cin >> input) {
+ client_interact->result = input.data();
+ client_interact->len = input.size();
+ }
+}
+
+std::auto_ptr<SecurityLayer> CyrusSasl::getSecurityLayer(uint16_t maxFrameSize)
+{
+ const void* value(0);
+ int result = sasl_getprop(conn, SASL_SSF, &value);
+ if (result != SASL_OK) {
+ throw framing::InternalErrorException(QPID_MSG("SASL error: " << sasl_errdetail(conn)));
+ }
+ uint ssf = *(reinterpret_cast<const unsigned*>(value));
+ std::auto_ptr<SecurityLayer> securityLayer;
+ if (ssf) {
+ QPID_LOG(info, "Installing security layer, SSF: "<< ssf);
+ securityLayer = std::auto_ptr<SecurityLayer>(new CyrusSecurityLayer(conn, maxFrameSize));
+ }
+ return securityLayer;
+}
+
+int getLogin(void* context, int /*id*/, const char** result, unsigned* /*len*/)
+{
+ if (context) {
+ char* login = (char*) context;
+ int status = getlogin_r(login, MAX_LOGIN_LENGTH);
+ if (status == 0) {
+ *result = login;
+ QPID_LOG(debug, "getLogin(): " << (*result));
+ } else {
+ strcpy(login, "guest");
+ QPID_LOG(error, "getlogin_r() failed with " << status << "; defaulting to " << login);
+ }
+ return SASL_OK;
+ } else {
+ return SASL_FAIL;
+ }
+}
+
+int getUserFromSettings(void* context, int /*id*/, const char** result, unsigned* /*len*/)
+{
+ if (context) {
+ *result = ((ConnectionSettings*) context)->username.c_str();
+ QPID_LOG(debug, "getUserFromSettings(): " << (*result));
+ return SASL_OK;
+ } else {
+ return SASL_FAIL;
+ }
+}
+
+int getPasswordFromSettings(sasl_conn_t* /*conn*/, void* context, int /*id*/, sasl_secret_t** psecret)
+{
+ if (context) {
+ size_t length = ((ConnectionSettings*) context)->password.size();
+ sasl_secret_t* secret = (sasl_secret_t*) malloc(sizeof(sasl_secret_t) + length);
+ secret->len = length;
+ memcpy(secret->data, ((ConnectionSettings*) context)->password.data(), length);
+ *psecret = secret;
+ return SASL_OK;
+ } else {
+ return SASL_FAIL;
+ }
+}
+
+}} // namespace qpid::client
+
+#endif
Added: qpid/trunk/qpid/cpp/src/qpid/client/SaslFactory.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/SaslFactory.h?rev=732082&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/SaslFactory.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/client/SaslFactory.h Tue Jan 6 11:50:59 2009
@@ -0,0 +1,48 @@
+#ifndef QPID_CLIENT_SASLFACTORY_H
+#define QPID_CLIENT_SASLFACTORY_H
+
+/*
+ *
+ * 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.
+ *
+ */
+#include "Sasl.h"
+#include "qpid/sys/Mutex.h"
+#include <memory>
+
+namespace qpid {
+namespace client {
+
+/**
+ * Factory for instances of the Sasl interface through which Sasl
+ * support is provided to a ConnectionHandler.
+ */
+class SaslFactory
+{
+ public:
+ std::auto_ptr<Sasl> create(const ConnectionSettings&);
+ static SaslFactory& getInstance();
+ ~SaslFactory();
+ private:
+ SaslFactory();
+ static qpid::sys::Mutex lock;
+ static std::auto_ptr<SaslFactory> instance;
+};
+}} // namespace qpid::client
+
+#endif /*!QPID_CLIENT_SASLFACTORY_H*/
Modified: qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp Tue Jan 6 11:50:59 2009
@@ -184,16 +184,21 @@
return;
}
if (codec == 0) return;
- if (codec->canEncode()) {
- // Try and get a queued buffer if not then construct new one
- AsynchIO::BufferBase* buff = aio->getQueuedBuffer();
- if (!buff) buff = new Buff;
- size_t encoded=codec->encode(buff->bytes, buff->byteCount);
- buff->dataCount = encoded;
- aio->queueWrite(buff);
- }
- if (codec->isClosed())
+ try {
+ if (codec->canEncode()) {
+ // Try and get a queued buffer if not then construct new one
+ AsynchIO::BufferBase* buff = aio->getQueuedBuffer();
+ if (!buff) buff = new Buff;
+ size_t encoded=codec->encode(buff->bytes, buff->byteCount);
+ buff->dataCount = encoded;
+ aio->queueWrite(buff);
+ }
+ if (codec->isClosed())
+ aio->queueWriteClose();
+ } catch (const std::exception& e) {
+ QPID_LOG(error, e.what());
aio->queueWriteClose();
+ }
}
}} // namespace qpid::sys
Added: qpid/trunk/qpid/cpp/src/qpid/sys/Codec.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/Codec.h?rev=732082&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/Codec.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/Codec.h Tue Jan 6 11:50:59 2009
@@ -0,0 +1,52 @@
+#ifndef QPID_SYS_CODEC_H
+#define QPID_SYS_CODEC_H
+
+/*
+ *
+ * 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.
+ *
+ */
+#include <cstddef>
+
+namespace qpid {
+namespace sys {
+
+/**
+ * Generic codec interface
+ */
+class Codec
+{
+ public:
+ virtual ~Codec() {}
+
+ /** Decode from buffer, return number of bytes decoded.
+ * @return may be less than size if there was incomplete
+ * data at the end of the buffer.
+ */
+ virtual size_t decode(const char* buffer, size_t size) = 0;
+
+
+ /** Encode into buffer, return number of bytes encoded */
+ virtual size_t encode(const char* buffer, size_t size) = 0;
+
+ /** Return true if we have data to encode */
+ virtual bool canEncode() = 0;
+};
+}} // namespace qpid::sys
+
+#endif /*!QPID_SYS_CODEC_H*/
Modified: qpid/trunk/qpid/cpp/src/qpid/sys/ConnectionCodec.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/ConnectionCodec.h?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/ConnectionCodec.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/ConnectionCodec.h Tue Jan 6 11:50:59 2009
@@ -21,6 +21,7 @@
* under the License.
*
*/
+#include "Codec.h"
#include "qpid/framing/ProtocolVersion.h"
namespace qpid {
@@ -34,23 +35,10 @@
* Interface of coder/decoder for a connection of a specific protocol
* version.
*/
-class ConnectionCodec {
+class ConnectionCodec : public Codec {
public:
virtual ~ConnectionCodec() {}
- /** Decode from buffer, return number of bytes decoded.
- * @return may be less than size if there was incomplete
- * data at the end of the buffer.
- */
- virtual size_t decode(const char* buffer, size_t size) = 0;
-
-
- /** Encode into buffer, return number of bytes encoded */
- virtual size_t encode(const char* buffer, size_t size) = 0;
-
- /** Return true if we have data to encode */
- virtual bool canEncode() = 0;
-
/** Network connection was closed from other end. */
virtual void closed() = 0;
Added: qpid/trunk/qpid/cpp/src/qpid/sys/SecurityLayer.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/SecurityLayer.h?rev=732082&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/SecurityLayer.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/SecurityLayer.h Tue Jan 6 11:50:59 2009
@@ -0,0 +1,42 @@
+#ifndef QPID_SYS_SECURITYLAYER_H
+#define QPID_SYS_SECURITYLAYER_H
+
+/*
+ *
+ * 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.
+ *
+ */
+#include "Codec.h"
+
+namespace qpid {
+namespace sys {
+
+/**
+ * Defines interface to a SASL negotiated Security Layer (for
+ * encryption/integrity)
+ */
+class SecurityLayer : public Codec
+{
+ public:
+ virtual void init(Codec*) = 0;
+ virtual ~SecurityLayer() {}
+};
+
+}} // namespace qpid::sys
+
+#endif /*!QPID_SYS_SECURITYLAYER_H*/
Modified: qpid/trunk/qpid/cpp/src/qpid/sys/TCPIOPlugin.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/TCPIOPlugin.cpp?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/TCPIOPlugin.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/TCPIOPlugin.cpp Tue Jan 6 11:50:59 2009
@@ -65,14 +65,10 @@
// Only provide to a Broker
if (broker) {
const broker::Broker::Options& opts = broker->getOptions();
- if (opts.requireEncrypted) {
- QPID_LOG(info, "Not accepting unencrypted connections on TCP");
- } else {
- ProtocolFactory::shared_ptr protocol(new AsynchIOProtocolFactory(opts.port, opts.connectionBacklog,
- opts.tcpNoDelay));
- QPID_LOG(notice, "Listening on TCP port " << protocol->getPort());
- broker->registerProtocolFactory("tcp", protocol);
- }
+ ProtocolFactory::shared_ptr protocol(new AsynchIOProtocolFactory(opts.port, opts.connectionBacklog,
+ opts.tcpNoDelay));
+ QPID_LOG(notice, "Listening on TCP port " << protocol->getPort());
+ broker->registerProtocolFactory("tcp", protocol);
}
}
} tcpPlugin;
Added: qpid/trunk/qpid/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.cpp?rev=732082&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.cpp Tue Jan 6 11:50:59 2009
@@ -0,0 +1,119 @@
+/*
+ *
+ * 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.
+ *
+ */
+#include "CyrusSecurityLayer.h"
+#include <algorithm>
+#include "qpid/framing/reply_exceptions.h"
+#include "qpid/log/Statement.h"
+
+namespace qpid {
+namespace sys {
+namespace cyrus {
+
+CyrusSecurityLayer::CyrusSecurityLayer(sasl_conn_t* c, uint16_t maxFrameSize) :
+ conn(c), decrypted(0), decryptedSize(0), encrypted(0), encryptedSize(0), codec(0), maxInputSize(0), decodeBuffer(maxFrameSize)
+{
+ const void* value(0);
+ int result = sasl_getprop(conn, SASL_MAXOUTBUF, &value);
+ if (result != SASL_OK) {
+ throw framing::InternalErrorException(QPID_MSG("SASL encode error: " << sasl_errdetail(conn)));
+ }
+ maxInputSize = *(reinterpret_cast<const unsigned*>(value));
+}
+
+size_t CyrusSecurityLayer::decode(const char* input, size_t size)
+{
+ size_t inStart = 0;
+ do {
+ size_t inSize = std::min(size - inStart, maxInputSize);
+ int result = sasl_decode(conn, input + inStart, inSize, &decrypted, &decryptedSize);
+ if (result != SASL_OK) {
+ throw framing::InternalErrorException(QPID_MSG("SASL encode error: " << sasl_errdetail(conn)));
+ }
+ inStart += inSize;
+ size_t copied = 0;
+ do {
+ size_t count = std::min(decryptedSize - copied, decodeBuffer.size - decodeBuffer.position);
+ ::memcpy(decodeBuffer.data + decodeBuffer.position, decrypted + copied, count);
+ copied += count;
+ decodeBuffer.position += count;
+ size_t decodedSize = codec->decode(decodeBuffer.data, decodeBuffer.position);
+ if (decodedSize < decodeBuffer.position) {
+ ::memmove(decodeBuffer.data, decodeBuffer.data + decodedSize, decodeBuffer.position - decodedSize);
+ }
+ decodeBuffer.position -= decodedSize;
+ } while (copied < decryptedSize);
+ } while (inStart < size);
+ return size;
+}
+
+size_t CyrusSecurityLayer::encode(const char* buffer, size_t size)
+{
+ size_t processed = 0;//records how many bytes have been written to buffer
+ do {
+ if (!encrypted) {
+ DataBuffer encodeBuffer(maxInputSize);//make sure maxInputSize > maxFrameSize
+ size_t encoded = codec->encode(encodeBuffer.data, encodeBuffer.size);
+ if (!encoded) break;//nothing more to do
+ int result = sasl_encode(conn, encodeBuffer.data, encoded, &encrypted, &encryptedSize);
+ if (result != SASL_OK) {
+ throw framing::InternalErrorException(QPID_MSG("SASL encode error: " << sasl_errdetail(conn)));
+ }
+ }
+ size_t remaining = size - processed;
+ if (remaining < encryptedSize) {
+ //can't fit all encrypted data in the buffer we've
+ //been given, copy in what we can and hold on to the
+ //rest until the next call
+ ::memcpy(const_cast<char*>(buffer + processed), encrypted, remaining);
+ processed += remaining;
+ encrypted += remaining;
+ encryptedSize -= remaining;
+ } else {
+ ::memcpy(const_cast<char*>(buffer + processed), encrypted, encryptedSize);
+ processed += encryptedSize;
+ encrypted = 0;
+ encryptedSize = 0;
+ }
+ } while (processed < size);
+ return processed;
+}
+
+bool CyrusSecurityLayer::canEncode()
+{
+ return encrypted || codec->canEncode();
+}
+
+void CyrusSecurityLayer::init(qpid::sys::Codec* c)
+{
+ codec = c;
+}
+
+CyrusSecurityLayer::DataBuffer::DataBuffer(size_t s) : position(0), size(s)
+{
+ data = new char[size];
+}
+
+CyrusSecurityLayer::DataBuffer::~DataBuffer()
+{
+ delete[] data;
+}
+
+}}} // namespace qpid::sys::cyrus
Added: qpid/trunk/qpid/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.h?rev=732082&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.h Tue Jan 6 11:50:59 2009
@@ -0,0 +1,66 @@
+#ifndef QPID_SYS_CYRUS_CYRUSSECURITYLAYER_H
+#define QPID_SYS_CYRUS_CYRUSSECURITYLAYER_H
+
+/*
+ *
+ * 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.
+ *
+ */
+
+#include "qpid/sys/IntegerTypes.h"
+#include "qpid/sys/SecurityLayer.h"
+#include <sasl/sasl.h>
+
+namespace qpid {
+namespace sys {
+namespace cyrus {
+
+
+/**
+ * Implementation of SASL security layer using cyrus-sasl library
+ */
+class CyrusSecurityLayer : public qpid::sys::SecurityLayer
+{
+ public:
+ CyrusSecurityLayer(sasl_conn_t*, uint16_t maxFrameSize);
+ size_t decode(const char* buffer, size_t size);
+ size_t encode(const char* buffer, size_t size);
+ bool canEncode();
+ void init(qpid::sys::Codec*);
+ private:
+ struct DataBuffer
+ {
+ char* data;
+ size_t position;
+ const size_t size;
+ DataBuffer(size_t);
+ ~DataBuffer();
+ };
+
+ sasl_conn_t* conn;
+ const char* decrypted;
+ unsigned decryptedSize;
+ const char* encrypted;
+ unsigned encryptedSize;
+ qpid::sys::Codec* codec;
+ size_t maxInputSize;
+ DataBuffer decodeBuffer;
+};
+}}} // namespace qpid::sys::cyrus
+
+#endif /*!QPID_SYS_CYRUS_CYRUSSECURITYLAYER_H*/
Modified: qpid/trunk/qpid/cpp/src/tests/.valgrind.supp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/.valgrind.supp?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/.valgrind.supp (original)
+++ qpid/trunk/qpid/cpp/src/tests/.valgrind.supp Tue Jan 6 11:50:59 2009
@@ -201,3 +201,10 @@
fun:_ZN5boost6detail3tss3setEPv
}
+{
+ Shows up on RHEL5: believed benign
+ Memcheck:Cond
+ fun:__strcpy_chk
+ fun:_sasl_load_plugins
+ fun:sasl_client_init
+}
Modified: qpid/trunk/qpid/cpp/src/tests/ConnectionOptions.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/ConnectionOptions.h?rev=732082&r1=732081&r2=732082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/ConnectionOptions.h (original)
+++ qpid/trunk/qpid/cpp/src/tests/ConnectionOptions.h Tue Jan 6 11:50:59 2009
@@ -47,7 +47,10 @@
("max-frame-size", optValue(maxFrameSize, "N"), "the maximum frame size to request.")
("bounds-multiplier", optValue(bounds, "N"),
"bound size of write queue (as a multiple of the max frame size).")
- ("tcp-nodelay", optValue(tcpNoDelay), "Turn on tcp-nodelay");
+ ("tcp-nodelay", optValue(tcpNoDelay), "Turn on tcp-nodelay")
+ ("service", optValue(service, "SERVICE-NAME"), "SASL service name.")
+ ("min-ssf", optValue(minSsf, "N"), "Minimum acceptable strength for SASL security layer")
+ ("max-ssf", optValue(maxSsf, "N"), "Maximum acceptable strength for SASL security layer");
}
};