Cluster + Security

-----------------------------------

* initial observation of a problem was a 2% failure rate in perftests
  of 20,000 messages against a cluster with security enabled.
  Problem was occasional receit of encrypted frames before the
  security codec had been enabled.  This is fixed with locking in
  cluster code (no new locks in broker code) and a callback that is
  fired by broker::ConnectionHandler::Handler to tell the cluster
  code when the opening handshake has finished.
  This was never a problem in the non-clustered broker before because
  everything happened in a single thread.

* the brokers that "shadow" the connection must not have null
  authenticators rather than real ones, so that they go through all
  the motions but don't do anythig.  Only the directly-connected
  broker can perform the security handshake.

* once the directly-connected broker receives the real user ID
  from its callback, it mcasts that ID to all other brokers.
  Otherwise the shadowing brokers will al think that the user ID
  is "anonymous".
  Check this by doing a substantial perftest, and using
      qpid-stat -c localhost:PORT
  to confirm that the brokers all have the same userID for the
  same connection.

* the user ID, negotiated during the Sasl security startup, is
   communicated from the directly connected broker to all other
   cluster brokers.

* If security is *not* being used, then this code should *not* tell
  the brokers anything about the userID -- or it will step on the value
  that is being set by other code pathways.

* test program at cpp/src/tests/cluster_authentication_soak is not yet
  fully automated -- run it with something like
  "sudo ./cluster_authentication_soak 500"



git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@944158 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 76 insertions, 19 deletions

diff --git a/cpp/src/qpid/broker/Connection.cpp b/cpp/src/qpid/broker/Connection.cpp
index 2bb68b9f2d..51615e5b5f 100644
--- a/cpp/src/qpid/broker/Connection.cpp
+++ b/cpp/src/qpid/broker/Connection.cpp
@@ -39,6 +39,8 @@
 #include <iostream>
 #include <assert.h>
 
+
+
 using namespace qpid::sys;
 using namespace qpid::framing;
 using qpid::ptr_map_ptr;
@@ -77,7 +79,7 @@ Connection::Connection(ConnectionOutputHandler* out_, Broker& broker_, const std
                        const qpid::sys::SecuritySettings& external, bool isLink_, uint64_t objectId, bool shadow_) :
     ConnectionState(out_, broker_),
     securitySettings(external),
-    adapter(*this, isLink_),
+    adapter(*this, isLink_, shadow_),
     isLink(isLink_),
     mgmtClosing(false),
     mgmtId(mgmtId_),
@@ -384,4 +386,7 @@ void Connection::restartTimeout()
         timeoutTimer->touch();
 }
 
+
+
+
 }}
diff --git a/cpp/src/qpid/broker/Connection.h b/cpp/src/qpid/broker/Connection.h
index 30a763411f..0639bcbb42 100644
--- a/cpp/src/qpid/broker/Connection.h
+++ b/cpp/src/qpid/broker/Connection.h
@@ -63,6 +63,9 @@ class LinkRegistry;
 class SecureConnection;
 struct ConnectionTimeoutTask;
 
+typedef boost::function<void ( std::string& )> userIdCallback;
+
+
 class Connection : public sys::ConnectionInputHandler,
                    public ConnectionState,
                    public RefCounted
@@ -143,6 +146,10 @@ class Connection : public sys::ConnectionInputHandler,
         return securitySettings;
     }
 
+    void setUserIdCallback ( UserIdCallback fn ) {
+        adapter.setUserIdCallback ( fn );
+    }
+
   private:
     typedef boost::ptr_map<framing::ChannelId, SessionHandler> ChannelMap;
     typedef std::vector<Queue::shared_ptr>::iterator queue_iterator;
diff --git a/cpp/src/qpid/broker/ConnectionHandler.cpp b/cpp/src/qpid/broker/ConnectionHandler.cpp
index 50a5aff2c9..b2d4210473 100644
--- a/cpp/src/qpid/broker/ConnectionHandler.cpp
+++ b/cpp/src/qpid/broker/ConnectionHandler.cpp
@@ -83,11 +83,11 @@ void ConnectionHandler::setSecureConnection(SecureConnection* secured)
     handler->secured = secured;
 }
 
-ConnectionHandler::ConnectionHandler(Connection& connection, bool isClient)  : handler(new Handler(connection, isClient)) {}
+ConnectionHandler::ConnectionHandler(Connection& connection, bool isClient, bool isShadow)  : handler(new Handler(connection, isClient, isShadow)) {}
 
-ConnectionHandler::Handler::Handler(Connection& c, bool isClient) :
+ConnectionHandler::Handler::Handler(Connection& c, bool isClient, bool isShadow) :
     proxy(c.getOutput()),
-    connection(c), serverMode(!isClient), acl(0), secured(0)
+    connection(c), serverMode(!isClient), acl(0), secured(0), userIdCallback(0)
 {
     if (serverMode) {
 
@@ -98,7 +98,7 @@ ConnectionHandler::Handler::Handler(Connection& c, bool isClient) :
 
         properties.setString(QPID_FED_TAG, connection.getBroker().getFederationTag());
 
-        authenticator = SaslAuthenticator::createAuthenticator(c);
+        authenticator = SaslAuthenticator::createAuthenticator(c, isShadow);
         authenticator->getMechanisms(mechanisms);
 
         Array locales(0x95);
@@ -181,6 +181,14 @@ void ConnectionHandler::Handler::tuneOk(uint16_t /*channelmax*/,
     connection.setHeartbeatInterval(heartbeat);
 }
 
+void ConnectionHandler::Handler::callUserIdCallbacks ( ) {
+    string s;
+    if ( false == authenticator->getUsername(s) )
+        s = "none";
+    if ( userIdCallback )
+      userIdCallback ( s );
+}
+
 void ConnectionHandler::Handler::open(const string& /*virtualHost*/,
                                       const framing::Array& /*capabilities*/, bool /*insist*/)
 {
@@ -195,6 +203,8 @@ void ConnectionHandler::Handler::open(const string& /*virtualHost*/,
         std::auto_ptr<SecurityLayer> sl = authenticator->getSecurityLayer(connection.getFrameMax());
         if (sl.get()) secured->activateSecurityLayer(sl);
     }
+
+    callUserIdCallbacks ( );
 }
 
 
diff --git a/cpp/src/qpid/broker/ConnectionHandler.h b/cpp/src/qpid/broker/ConnectionHandler.h
index d74f65da36..0372942188 100644
--- a/cpp/src/qpid/broker/ConnectionHandler.h
+++ b/cpp/src/qpid/broker/ConnectionHandler.h
@@ -40,6 +40,9 @@ namespace broker {
 class Connection;
 class SecureConnection;
 
+typedef boost::function<void ( std::string& )> UserIdCallback;
+
+
 class ConnectionHandler : public framing::FrameHandler
 {
     struct Handler : public framing::AMQP_AllOperations::ConnectionHandler
@@ -51,7 +54,7 @@ class ConnectionHandler : public framing::FrameHandler
         AclModule* acl;
         SecureConnection* secured;
 
-        Handler(Connection& connection, bool isClient);
+        Handler(Connection& connection, bool isClient, bool isShadow=false);
         ~Handler();
         void startOk(const qpid::framing::FieldTable& clientProperties,
                      const std::string& mechanism, const std::string& response,
@@ -64,6 +67,14 @@ class ConnectionHandler : public framing::FrameHandler
         void close(uint16_t replyCode, const std::string& replyText);
         void closeOk();
 
+        UserIdCallback userIdCallback;
+        void setUserIdCallback ( UserIdCallback fn ) {
+                 userIdCallback = fn;
+             };
+
+
+        void callUserIdCallbacks ( );
+
 
         void start(const qpid::framing::FieldTable& serverProperties,
                    const framing::Array& mechanisms,
@@ -81,12 +92,17 @@ class ConnectionHandler : public framing::FrameHandler
         void redirect(const std::string& host, const framing::Array& knownHosts);
     };
     std::auto_ptr<Handler> handler;
+
+
   public:
-    ConnectionHandler(Connection& connection, bool isClient);
+    ConnectionHandler(Connection& connection, bool isClient, bool isShadow=false );
     void close(framing::connection::CloseCode code, const std::string& text);
     void heartbeat();
     void handle(framing::AMQFrame& frame);
     void setSecureConnection(SecureConnection* secured);
+    void setUserIdCallback ( UserIdCallback fn ) {
+      handler->setUserIdCallback ( fn );
+    }
 };
 
 
diff --git a/cpp/src/qpid/broker/SaslAuthenticator.cpp b/cpp/src/qpid/broker/SaslAuthenticator.cpp
index 0f72f9643d..c55f3edb38 100644
--- a/cpp/src/qpid/broker/SaslAuthenticator.cpp
+++ b/cpp/src/qpid/broker/SaslAuthenticator.cpp
@@ -41,10 +41,12 @@ using qpid::sys::SecuritySettings;
 using boost::format;
 using boost::str;
 
+
 namespace qpid {
 namespace broker {
 
 
+
 class NullAuthenticator : public SaslAuthenticator
 {
     Connection& connection;
@@ -62,6 +64,8 @@ public:
 
 #if HAVE_SASL
 
+
+
 class CyrusAuthenticator : public SaslAuthenticator
 {
     sasl_conn_t *sasl_conn;
@@ -84,8 +88,7 @@ public:
     std::auto_ptr<SecurityLayer> getSecurityLayer(uint16_t maxFrameSize);
 };
 
-bool SaslAuthenticator::available(void)
-{
+bool SaslAuthenticator::available(void) {
     return true;
 }
 
@@ -109,8 +112,7 @@ void SaslAuthenticator::fini(void)
 
 typedef NullAuthenticator CyrusAuthenticator;
 
-bool SaslAuthenticator::available(void)
-{
+bool SaslAuthenticator::available(void) {
     return false;
 }
 
@@ -126,18 +128,20 @@ void SaslAuthenticator::fini(void)
 
 #endif
 
-std::auto_ptr<SaslAuthenticator> SaslAuthenticator::createAuthenticator(Connection& c)
+std::auto_ptr<SaslAuthenticator> SaslAuthenticator::createAuthenticator(Connection& c, bool isShadow )
 {
-    static bool needWarning = true;
     if (c.getBroker().getOptions().auth) {
-        return std::auto_ptr<SaslAuthenticator>(new CyrusAuthenticator(c, c.getBroker().getOptions().requireEncrypted));
+        if ( isShadow )
+            return std::auto_ptr<SaslAuthenticator>(new NullAuthenticator(c, c.getBroker().getOptions().requireEncrypted));
+        else 
+            return std::auto_ptr<SaslAuthenticator>(new CyrusAuthenticator(c, c.getBroker().getOptions().requireEncrypted));
     } else {
         QPID_LOG(debug, "SASL: No Authentication Performed");
-        needWarning = false;
         return std::auto_ptr<SaslAuthenticator>(new NullAuthenticator(c, c.getBroker().getOptions().requireEncrypted));
     }
 }
 
+
 NullAuthenticator::NullAuthenticator(Connection& c, bool e) : connection(c), client(c.getOutput()), 
                                                               realm(c.getBroker().getOptions().realm), encrypt(e) {}
 NullAuthenticator::~NullAuthenticator() {}
@@ -200,7 +204,6 @@ std::auto_ptr<SecurityLayer> NullAuthenticator::getSecurityLayer(uint16_t)
 
 #if HAVE_SASL
 
-
 CyrusAuthenticator::CyrusAuthenticator(Connection& c, bool _encrypt) : 
     sasl_conn(0), connection(c), client(c.getOutput()), encrypt(_encrypt)
 {
@@ -386,7 +389,7 @@ void CyrusAuthenticator::processAuthenticationStep(int code, const char *challen
             // authentication failure, when one is available
             throw ConnectionForcedException("Authenticated username unavailable");
         }
-        QPID_LOG(info, "SASL: Authentication succeeded for: " << uid);
+        QPID_LOG(info, connection.getMgmtId() << " SASL: Authentication succeeded for: " << uid);
 
         connection.setUserId(uid);
 
@@ -432,7 +435,6 @@ std::auto_ptr<SecurityLayer> CyrusAuthenticator::getSecurityLayer(uint16_t maxFr
     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;
diff --git a/cpp/src/qpid/broker/SaslAuthenticator.h b/cpp/src/qpid/broker/SaslAuthenticator.h
index 8ddaeb19a4..f4ad24b3bd 100644
--- a/cpp/src/qpid/broker/SaslAuthenticator.h
+++ b/cpp/src/qpid/broker/SaslAuthenticator.h
@@ -21,17 +21,27 @@
 #ifndef _SaslAuthenticator_
 #define _SaslAuthenticator_
 
+
 #include "qpid/framing/amqp_types.h"
 #include "qpid/framing/AMQP_ClientProxy.h"
 #include "qpid/Exception.h"
 #include "qpid/sys/SecurityLayer.h"
 #include <memory>
+#include <vector>
+#include <boost/bind.hpp>
+#include <boost/function.hpp>
 
 namespace qpid {
 namespace broker {
 
 class Connection;
 
+// Calls your fn with the user ID string, just 
+// after the security negotiation is complete.
+// Add your callback to the list with addUserIdCallback().
+typedef boost::function<void ( std::string& )> UserIdCallback;
+
+
 class SaslAuthenticator
 {
 public:
@@ -40,16 +50,23 @@ public:
     virtual void start(const std::string& mechanism, const std::string& response) = 0;
     virtual void step(const std::string& response) = 0;
     virtual void getUid(std::string&) {}
+    virtual bool getUsername(std::string&) { return false; };
     virtual void getError(std::string&) {}
     virtual std::auto_ptr<qpid::sys::SecurityLayer> getSecurityLayer(uint16_t maxFrameSize) = 0;
 
+    virtual void setUserIdCallback ( UserIdCallback ) { }
     static bool available(void);
 
     // Initialize the SASL mechanism; throw if it fails.
     static void init(const std::string& saslName);
     static void fini(void);
 
-    static std::auto_ptr<SaslAuthenticator> createAuthenticator(Connection& connection);
+    static std::auto_ptr<SaslAuthenticator> createAuthenticator(Connection& connection, bool isShadow);
+
+    virtual void callUserIdCallbacks() { }
+
+private:
+    UserIdCallback userIdCallback;
 };
 
 }}