SASLizing Interbroker Links

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

1. Brokers already knew how to handle the server side of SASLized
   links, but not the client side.  So we promoted the client-side
   SASL code from the client library to the common library so that
   the broker could also use it.  This affected SaslFactory.{h,cpp}
   and Sasl.h
   TODO -- can the server-side and client-side code be unified here?

2. Some of the SASL verbs in broker/ConnectionHandler.cpp are
   expanded: start, secure, tune.

3. broker/SecureConnection is altered to get the client-broker and
   the server-broker to agree on when the security layer should be
   inserted.

4. the python tool qpid-route is modified so that, in the "route add"
   command, you can specify the security mechanism for SASL to use.
   TODO -- should we also pass in {min,max}SSF ?

5. Changes in broker/LinkRegistry to allow the information input by
   qpid-route to be passed up to where it is needed.

6. A bash script test run by "make check" that creates a SASLized
   federation link and sends some messages down it.
   TODO - write a python unit test instead of a bash script.  I
          think I uncovered a bug in the python code when I tried.

7. NOTE - testing for this feature does not work with versions of
   SASL earlier than 2.1.22, becuase I can't tell SASL to use a
   SASL database file in a nonstandard location.  The test is
   disabled for earlier versions.




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

18 files changed, 424 insertions, 59 deletions

diff --git a/qpid/cpp/src/CMakeLists.txt b/qpid/cpp/src/CMakeLists.txt
index e52ccb6777..ba2a866cfb 100644
--- a/qpid/cpp/src/CMakeLists.txt
+++ b/qpid/cpp/src/CMakeLists.txt
@@ -712,7 +712,6 @@ else (CMAKE_SYSTEM_NAME STREQUAL Windows)
   )
 
   set (qpidclient_platform_SOURCES
-     qpid/client/SaslFactory.cpp
   )
 
   set (qpidd_platform_SOURCES
@@ -732,6 +731,7 @@ set (qpidcommon_SOURCES
      qpid/Options.cpp
      qpid/Plugin.cpp
      qpid/RefCountedBuffer.cpp
+     qpid/SaslFactory.cpp
      qpid/SessionState.cpp
      qpid/SessionId.cpp
      qpid/StringUtils.cpp
diff --git a/qpid/cpp/src/Makefile.am b/qpid/cpp/src/Makefile.am
index 0ce1480825..9a6f7f3949 100644
--- a/qpid/cpp/src/Makefile.am
+++ b/qpid/cpp/src/Makefile.am
@@ -331,6 +331,9 @@ libqpidcommon_la_SOURCES +=			\
   qpid/RefCounted.h				\
   qpid/RefCountedBuffer.cpp			\
   qpid/RefCountedBuffer.h			\
+  qpid/Sasl.h                                   \
+  qpid/SaslFactory.cpp                          \
+  qpid/SaslFactory.h                            \
   qpid/Serializer.h				\
   qpid/SessionId.cpp				\
   qpid/SessionState.cpp				\
@@ -692,9 +695,6 @@ libqpidclient_la_SOURCES =			\
   qpid/client/QueueOptions.cpp			\
   qpid/client/Results.cpp			\
   qpid/client/Results.h				\
-  qpid/client/Sasl.h				\
-  qpid/client/SaslFactory.cpp			\
-  qpid/client/SaslFactory.h			\
   qpid/client/SessionBase_0_10.cpp		\
   qpid/client/SessionBase_0_10Access.h		\
   qpid/client/SessionImpl.cpp			\
diff --git a/qpid/cpp/src/qpid/client/Sasl.h b/qpid/cpp/src/qpid/Sasl.h
index 56735a5fc3..9a9d61b037 100644
--- a/qpid/cpp/src/qpid/client/Sasl.h
+++ b/qpid/cpp/src/qpid/Sasl.h
@@ -1,5 +1,5 @@
-#ifndef QPID_CLIENT_SASL_H
-#define QPID_CLIENT_SASL_H
+#ifndef QPID_SASL_H
+#define QPID_SASL_H
 
 /*
  *
@@ -33,10 +33,6 @@ class SecurityLayer;
 struct SecuritySettings;
 }
 
-namespace client {
-
-struct ConnectionSettings;
-
 /**
  * Interface to SASL support. This class is implemented by platform-specific
  * SASL providers.
@@ -59,6 +55,6 @@ class Sasl
     virtual std::auto_ptr<qpid::sys::SecurityLayer> getSecurityLayer(uint16_t maxFrameSize) = 0;    
     virtual ~Sasl() {}
 };
-}} // namespace qpid::client
+} // namespace qpid
 
-#endif  /*!QPID_CLIENT_SASL_H*/
+#endif  /*!QPID_SASL_H*/
diff --git a/qpid/cpp/src/qpid/client/SaslFactory.cpp b/qpid/cpp/src/qpid/SaslFactory.cpp
index 79acf3cd7c..28c27f7529 100644
--- a/qpid/cpp/src/qpid/client/SaslFactory.cpp
+++ b/qpid/cpp/src/qpid/SaslFactory.cpp
@@ -18,8 +18,7 @@
  * under the License.
  *
  */
-#include "qpid/client/SaslFactory.h"
-#include "qpid/client/ConnectionSettings.h"
+#include "qpid//SaslFactory.h"
 #include <map>
 #include <string.h>
 
@@ -30,7 +29,6 @@
 #ifndef HAVE_SASL
 
 namespace qpid {
-namespace client {
 
 //Null implementation
 
@@ -47,7 +45,7 @@ SaslFactory& SaslFactory::getInstance()
     return *instance;
 }
 
-std::auto_ptr<Sasl> SaslFactory::create(const ConnectionSettings&)
+std::auto_ptr<Sasl> SaslFactory::create( const std::string &, const std::string &, const std::string &, const std::string &, int, int )
 {
     return std::auto_ptr<Sasl>();
 }
@@ -55,7 +53,7 @@ std::auto_ptr<Sasl> SaslFactory::create(const ConnectionSettings&)
 qpid::sys::Mutex SaslFactory::lock;
 std::auto_ptr<SaslFactory> SaslFactory::instance;
 
-}} // namespace qpid::client
+} // namespace qpid
 
 #else
 
@@ -69,7 +67,6 @@ std::auto_ptr<SaslFactory> SaslFactory::instance;
 #include <strings.h>
 
 namespace qpid {
-namespace client {
 
 using qpid::sys::SecurityLayer;
 using qpid::sys::SecuritySettings;
@@ -78,10 +75,42 @@ using qpid::framing::InternalErrorException;
 
 const size_t MAX_LOGIN_LENGTH = 50;
 
+struct CyrusSaslSettings
+{
+    CyrusSaslSettings ( ) :
+        username ( std::string(0) ),
+        password ( std::string(0) ),
+        service  ( std::string(0) ),
+        host     ( std::string(0) ),
+        minSsf ( 0 ),
+        maxSsf ( 0 )
+    {
+    }
+
+    CyrusSaslSettings ( const std::string & user, const std::string & password, const std::string & service, const std::string & host, int minSsf, int maxSsf ) :
+        username(user),
+        password(password),
+        service(service),
+        host(host),
+        minSsf(minSsf),
+        maxSsf(maxSsf)
+    {
+    }
+
+    std::string username,
+                password,
+                service,
+                host;
+
+    int minSsf,
+        maxSsf;
+};
+
+
 class CyrusSasl : public Sasl
 {
   public:
-    CyrusSasl(const ConnectionSettings&);
+    CyrusSasl(const std::string & username, const std::string & password, const std::string & serviceName, const std::string & hostName, int minSsf, int maxSsf);
     ~CyrusSasl();
     std::string start(const std::string& mechanisms, const SecuritySettings* externalSettings);
     std::string step(const std::string& challenge);
@@ -91,7 +120,7 @@ class CyrusSasl : public Sasl
   private:
     sasl_conn_t* conn;    
     sasl_callback_t callbacks[5];//realm, user, authname, password, end-of-list
-    ConnectionSettings settings;
+    CyrusSaslSettings settings;
     std::string input;
     std::string mechanism;
     char login[MAX_LOGIN_LENGTH];
@@ -130,13 +159,14 @@ SaslFactory& SaslFactory::getInstance()
     return *instance;
 }
 
-std::auto_ptr<Sasl> SaslFactory::create(const ConnectionSettings& settings)
+std::auto_ptr<Sasl> SaslFactory::create(const std::string & username, const std::string & password, const std::string & serviceName, const std::string & hostName, int minSsf, int maxSsf)
 {
-    std::auto_ptr<Sasl> sasl(new CyrusSasl(settings));
+    std::auto_ptr<Sasl> sasl(new CyrusSasl(username, password, serviceName, hostName, minSsf, maxSsf));
     return sasl;
 }
 
-CyrusSasl::CyrusSasl(const ConnectionSettings& s) : conn(0), settings(s) 
+CyrusSasl::CyrusSasl(const std::string & username, const std::string & password, const std::string & serviceName, const std::string & hostName, int minSsf, int maxSsf)
+    : conn(0), settings(username, password, serviceName, hostName, minSsf, maxSsf) 
 {
     size_t i = 0;
 
@@ -336,7 +366,7 @@ std::auto_ptr<SecurityLayer> CyrusSasl::getSecurityLayer(uint16_t maxFrameSize)
 int getUserFromSettings(void* context, int /*id*/, const char** result, unsigned* /*len*/)
 {
     if (context) {
-        *result = ((ConnectionSettings*) context)->username.c_str();
+        *result = ((CyrusSaslSettings*) context)->username.c_str();
         QPID_LOG(debug, "getUserFromSettings(): " << (*result));
         return SASL_OK;
     } else {
@@ -345,7 +375,7 @@ int getUserFromSettings(void* context, int /*id*/, const char** result, unsigned
 }
 
 namespace {
-// Global map of secrest allocated for SASL connections via callback
+// Global map of secrets allocated for SASL connections via callback
 // to getPasswordFromSettings. Ensures secrets are freed.
 class SecretsMap {
     typedef std::map<sasl_conn_t*, void*> Map;
@@ -368,11 +398,11 @@ SecretsMap getPasswordFromSettingsSecrets;
 int getPasswordFromSettings(sasl_conn_t* conn, void* context, int /*id*/, sasl_secret_t** psecret)
 {
     if (context) {
-        size_t length = ((ConnectionSettings*) context)->password.size();
+        size_t length = ((CyrusSaslSettings*) context)->password.size();
         sasl_secret_t* secret = (sasl_secret_t*) malloc(sizeof(sasl_secret_t) + length);
         getPasswordFromSettingsSecrets.keep(conn, secret);
         secret->len = length;
-        memcpy(secret->data, ((ConnectionSettings*) context)->password.data(), length);
+        memcpy(secret->data, ((CyrusSaslSettings*) context)->password.data(), length);
         *psecret = secret;
         return SASL_OK;
     } else {
@@ -380,6 +410,6 @@ int getPasswordFromSettings(sasl_conn_t* conn, void* context, int /*id*/, sasl_s
     }
 }
 
-}} // namespace qpid::client
+} // namespace qpid
 
 #endif
diff --git a/qpid/cpp/src/qpid/client/SaslFactory.h b/qpid/cpp/src/qpid/SaslFactory.h
index d012af06f7..d9d83c494d 100644
--- a/qpid/cpp/src/qpid/client/SaslFactory.h
+++ b/qpid/cpp/src/qpid/SaslFactory.h
@@ -1,5 +1,5 @@
-#ifndef QPID_CLIENT_SASLFACTORY_H
-#define QPID_CLIENT_SASLFACTORY_H
+#ifndef QPID_SASLFACTORY_H
+#define QPID_SASLFACTORY_H
 
 /*
  *
@@ -21,12 +21,11 @@
  * under the License.
  *
  */
-#include "qpid/client/Sasl.h"
+#include "qpid/Sasl.h"
 #include "qpid/sys/Mutex.h"
 #include <memory>
 
 namespace qpid {
-namespace client {
 
 /**
  * Factory for instances of the Sasl interface through which Sasl
@@ -35,7 +34,7 @@ namespace client {
 class SaslFactory
 {
   public:
-    std::auto_ptr<Sasl> create(const ConnectionSettings&);
+    std::auto_ptr<Sasl> create(const std::string & userName, const std::string & password, const std::string & serviceName, const std::string & hostName, int minSsf, int maxSsf );
     static SaslFactory& getInstance();
     ~SaslFactory();
   private:
@@ -43,6 +42,6 @@ class SaslFactory
     static qpid::sys::Mutex lock;
     static std::auto_ptr<SaslFactory> instance;
 };
-}} // namespace qpid::client
+} // namespace qpid
 
-#endif  /*!QPID_CLIENT_SASLFACTORY_H*/
+#endif  /*!QPID_SASLFACTORY_H*/
diff --git a/qpid/cpp/src/qpid/broker/Connection.cpp b/qpid/cpp/src/qpid/broker/Connection.cpp
index d50f0c946a..98743cdae6 100644
--- a/qpid/cpp/src/qpid/broker/Connection.cpp
+++ b/qpid/cpp/src/qpid/broker/Connection.cpp
@@ -192,6 +192,38 @@ string Connection::getAuthMechanism()
     return links.getAuthMechanism(mgmtId);
 }
 
+string Connection::getUsername ( )
+{
+    if (!isLink)
+        return string("anonymous");
+
+    return links.getUsername(mgmtId);
+}
+
+string Connection::getPassword ( )
+{
+    if (!isLink)
+        return string("");
+
+    return links.getPassword(mgmtId);
+}
+
+string Connection::getHost ( )
+{
+    if (!isLink)
+        return string("");
+
+    return links.getHost(mgmtId);
+}
+
+uint16_t Connection::getPort ( )
+{
+    if (!isLink)
+        return 0;
+
+    return links.getPort(mgmtId);
+}
+
 string Connection::getAuthCredentials()
 {
     if (!isLink)
diff --git a/qpid/cpp/src/qpid/broker/Connection.h b/qpid/cpp/src/qpid/broker/Connection.h
index c1b2b5a8fc..d978187e0c 100644
--- a/qpid/cpp/src/qpid/broker/Connection.h
+++ b/qpid/cpp/src/qpid/broker/Connection.h
@@ -115,6 +115,10 @@ class Connection : public sys::ConnectionInputHandler,
     void recordFromClient (framing::AMQFrame& frame);
     std::string getAuthMechanism();
     std::string getAuthCredentials();
+    std::string getUsername();
+    std::string getPassword();
+    std::string getHost();
+    uint16_t    getPort();
     void notifyConnectionForced(const std::string& text);
     void setUserId(const std::string& uid);
     void raiseConnectEvent();
diff --git a/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp b/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp
index c349bc7ac7..c812374d38 100644
--- a/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp
+++ b/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp
@@ -20,6 +20,7 @@
  *
  */
 
+#include "qpid/SaslFactory.h"
 #include "qpid/broker/ConnectionHandler.h"
 #include "qpid/broker/Connection.h"
 #include "qpid/broker/SecureConnection.h"
@@ -49,6 +50,7 @@ const std::string CLIENT_PROCESS_NAME("qpid.client_process");
 const std::string CLIENT_PID("qpid.client_pid");
 const std::string CLIENT_PPID("qpid.client_ppid");
 const int SESSION_FLOW_CONTROL_VER = 1;
+const std::string SPACE(" ");
 }
 
 void ConnectionHandler::close(connection::CloseCode code, const string& text)
@@ -106,7 +108,10 @@ ConnectionHandler::Handler::Handler(Connection& c, bool isClient, bool isShadow)
         boost::shared_ptr<FieldValue> l(new Str16Value(en_US));
         locales.add(l);
         proxy.start(properties, mechanisms, locales);
+        
     }
+
+    maxFrameSize = (64 * 1024) - 1;
 }
 
 
@@ -230,33 +235,105 @@ void ConnectionHandler::Handler::heartbeat(){
 }
 
 void ConnectionHandler::Handler::start(const FieldTable& serverProperties,
-                                       const framing::Array& /*mechanisms*/,
+                                       const framing::Array& supportedMechanisms,
                                        const framing::Array& /*locales*/)
 {
-    string mechanism = connection.getAuthMechanism();
+    string requestedMechanism = connection.getAuthMechanism();
     string response  = connection.getAuthCredentials();
 
+    std::string username = connection.getUsername();
+    std::string password = connection.getPassword();
+    std::string host     = connection.getHost();
+    std::string service("qpidd");
+
+    sasl = SaslFactory::getInstance().create( username,
+                                              password,
+                                              service,
+                                              host,
+                                              0,   // TODO -- mgoulish Fri Sep 24 06:41:26 EDT 2010
+                                              256  /* TODO -- mgoulish*/ );
+    std::string supportedMechanismsList;
+    bool requestedMechanismIsSupported = false;
+    Array::const_iterator i;
+
+    /*
+      If no specific mechanism has been requested, just make
+      a list of all of them, and assert that the one the caller
+      requested is there.  ( If *any* are supported! )
+    */
+    if ( requestedMechanism.empty() ) {
+        for ( i = supportedMechanisms.begin(); i != supportedMechanisms.end(); ++i) {
+            if (i != supportedMechanisms.begin())
+                supportedMechanismsList += SPACE;
+            supportedMechanismsList += (*i)->get<std::string>();
+            requestedMechanismIsSupported = true;
+        }
+    }
+    else {
+        requestedMechanismIsSupported = false;
+        /*
+          The caller has requested a mechanism.  If it's available,
+          make sure it ends up at the head of the list.
+        */
+        for ( i = supportedMechanisms.begin(); i != supportedMechanisms.end(); ++i) {
+            string currentMechanism = (*i)->get<std::string>();
+
+            if ( requestedMechanism == currentMechanism ) {
+                requestedMechanismIsSupported = true;
+                supportedMechanismsList = currentMechanism + SPACE + supportedMechanismsList;
+            } else {
+                if (i != supportedMechanisms.begin())
+                    supportedMechanismsList += SPACE;
+                supportedMechanismsList += currentMechanism;
+            }
+        }
+    }
+
     connection.setFederationPeerTag(serverProperties.getAsString(QPID_FED_TAG));
 
     FieldTable ft;
     ft.setInt(QPID_FED_LINK,1);
     ft.setString(QPID_FED_TAG, connection.getBroker().getFederationTag());
-    proxy.startOk(ft, mechanism, response, en_US);
+
+    if (sasl.get()) {
+        string response =
+            sasl->start ( requestedMechanism.empty()
+                              ? supportedMechanismsList
+                              : requestedMechanism,
+                          getSecuritySettings 
+                              ? getSecuritySettings()
+                              : 0
+                        );
+        proxy.startOk ( ft, sasl->getMechanism(), response, en_US );
+    }
+    else {
+        string response = ((char)0) + username + ((char)0) + password;
+        proxy.startOk ( ft, requestedMechanism, response, en_US );
+    }
+
 }
 
-void ConnectionHandler::Handler::secure(const string& /*challenge*/)
+void ConnectionHandler::Handler::secure(const string& challenge )
 {
-    proxy.secureOk("");
+    if (sasl.get()) {
+        string response = sasl->step(challenge);
+        proxy.secureOk(response);
+    }
+    else {
+        proxy.secureOk("");
+    }
 }
 
 void ConnectionHandler::Handler::tune(uint16_t channelMax,
-                                      uint16_t frameMax,
+                                      uint16_t maxFrameSizeProposed,
                                       uint16_t /*heartbeatMin*/,
                                       uint16_t heartbeatMax)
 {
-    connection.setFrameMax(frameMax);
+    maxFrameSize = std::min(maxFrameSize, maxFrameSizeProposed);
+    connection.setFrameMax(maxFrameSize);
+
     connection.setHeartbeat(heartbeatMax);
-    proxy.tuneOk(channelMax, frameMax, heartbeatMax);
+    proxy.tuneOk(channelMax, maxFrameSize, heartbeatMax);
     proxy.open("/", Array(), true);
 }
 
@@ -266,6 +343,17 @@ void ConnectionHandler::Handler::openOk(const framing::Array& knownHosts)
         Url url((*i)->get<std::string>());
         connection.getKnownHosts().push_back(url);
     }
+
+    if (sasl.get()) {
+        std::auto_ptr<qpid::sys::SecurityLayer> securityLayer = sasl->getSecurityLayer(maxFrameSize);
+
+        if ( securityLayer.get() ) {
+          secured->activateSecurityLayer(securityLayer, true);
+        }
+
+        saslUserId = sasl->getUserId();
+    }
+
     isOpen = true;
 }
 
diff --git a/qpid/cpp/src/qpid/broker/ConnectionHandler.h b/qpid/cpp/src/qpid/broker/ConnectionHandler.h
index 6d55cab647..70882a24e9 100644
--- a/qpid/cpp/src/qpid/broker/ConnectionHandler.h
+++ b/qpid/cpp/src/qpid/broker/ConnectionHandler.h
@@ -22,6 +22,7 @@
 #define _ConnectionAdapter_
 
 #include <memory>
+#include "qpid/Sasl.h"
 #include "qpid/broker/SaslAuthenticator.h"
 #include "qpid/framing/amqp_types.h"
 #include "qpid/framing/AMQFrame.h"
@@ -33,8 +34,16 @@
 #include "qpid/framing/ProtocolVersion.h"
 #include "qpid/Exception.h"
 #include "qpid/broker/AclModule.h"
+#include "qpid/sys/SecurityLayer.h"
+
 
 namespace qpid {
+
+namespace sys {
+struct SecuritySettings;
+}
+
+
 namespace broker {
 
 class Connection;
@@ -79,6 +88,12 @@ class ConnectionHandler : public framing::FrameHandler
         void openOk(const framing::Array& knownHosts);
 
         void redirect(const std::string& host, const framing::Array& knownHosts);
+
+        std::auto_ptr<Sasl> sasl;
+        typedef boost::function<const qpid::sys::SecuritySettings*()> GetSecuritySettings;
+        GetSecuritySettings  getSecuritySettings;     /* query the transport for its security details */
+        std::string saslUserId;
+        uint16_t maxFrameSize;
     };
     std::auto_ptr<Handler> handler;
 
diff --git a/qpid/cpp/src/qpid/broker/LinkRegistry.cpp b/qpid/cpp/src/qpid/broker/LinkRegistry.cpp
index 9d429a2dcc..ea14552cc1 100644
--- a/qpid/cpp/src/qpid/broker/LinkRegistry.cpp
+++ b/qpid/cpp/src/qpid/broker/LinkRegistry.cpp
@@ -312,6 +312,42 @@ std::string LinkRegistry::getAuthCredentials(const std::string& key)
     return result;
 }
 
+std::string LinkRegistry::getUsername(const std::string& key)
+{
+    Link::shared_ptr link = findLink(key);
+    if (!link)
+        return string();
+
+    return link->getUsername();
+}
+
+std::string LinkRegistry::getHost(const std::string& key)
+{
+    Link::shared_ptr link = findLink(key);
+    if (!link)
+        return string();
+
+    return link->getHost();
+}
+
+uint16_t LinkRegistry::getPort(const std::string& key)
+{
+    Link::shared_ptr link = findLink(key);
+    if (!link)
+        return 0;
+
+    return link->getPort();
+}
+
+std::string LinkRegistry::getPassword(const std::string& key)
+{
+    Link::shared_ptr link = findLink(key);
+    if (!link)
+        return string();
+
+    return link->getPassword();
+}
+
 std::string LinkRegistry::getAuthIdentity(const std::string& key)
 {
     Link::shared_ptr link = findLink(key);
diff --git a/qpid/cpp/src/qpid/broker/LinkRegistry.h b/qpid/cpp/src/qpid/broker/LinkRegistry.h
index 52ab700cfc..a1931920d7 100644
--- a/qpid/cpp/src/qpid/broker/LinkRegistry.h
+++ b/qpid/cpp/src/qpid/broker/LinkRegistry.h
@@ -132,6 +132,10 @@ namespace broker {
         std::string getAuthMechanism   (const std::string& key);
         std::string getAuthCredentials (const std::string& key);
         std::string getAuthIdentity    (const std::string& key);
+        std::string getUsername        (const std::string& key);
+        std::string getPassword        (const std::string& key);
+        std::string getHost            (const std::string& key);
+        uint16_t    getPort            (const std::string& key);
 
         /**
          * Called by links failing over to new address
diff --git a/qpid/cpp/src/qpid/broker/SecureConnection.cpp b/qpid/cpp/src/qpid/broker/SecureConnection.cpp
index 74aec239ca..5c1ebf3e8b 100644
--- a/qpid/cpp/src/qpid/broker/SecureConnection.cpp
+++ b/qpid/cpp/src/qpid/broker/SecureConnection.cpp
@@ -78,10 +78,13 @@ void SecureConnection:: setCodec(std::auto_ptr<ConnectionCodec> c)
     codec = c;
 }
 
-void SecureConnection::activateSecurityLayer(std::auto_ptr<SecurityLayer> sl)
+void SecureConnection::activateSecurityLayer(std::auto_ptr<SecurityLayer> sl, bool secureImmediately)
 {
     securityLayer = sl;
     securityLayer->init(codec.get());
+
+    if ( secureImmediately )
+        secured = true;
 }
 
 }} // namespace qpid::broker
diff --git a/qpid/cpp/src/qpid/broker/SecureConnection.h b/qpid/cpp/src/qpid/broker/SecureConnection.h
index 4a0cc50e34..1547faae1e 100644
--- a/qpid/cpp/src/qpid/broker/SecureConnection.h
+++ b/qpid/cpp/src/qpid/broker/SecureConnection.h
@@ -49,7 +49,7 @@ class SecureConnection : public qpid::sys::ConnectionCodec
     bool isClosed() const;
     framing::ProtocolVersion getVersion() const;
     void setCodec(std::auto_ptr<ConnectionCodec>);
-    void activateSecurityLayer(std::auto_ptr<qpid::sys::SecurityLayer>);
+    void activateSecurityLayer(std::auto_ptr<qpid::sys::SecurityLayer>, bool secureImmediately=false);
   private:
     std::auto_ptr<ConnectionCodec> codec;
     std::auto_ptr<qpid::sys::SecurityLayer> securityLayer;
diff --git a/qpid/cpp/src/qpid/client/ConnectionHandler.cpp b/qpid/cpp/src/qpid/client/ConnectionHandler.cpp
index e615878703..8dc1e8338a 100644
--- a/qpid/cpp/src/qpid/client/ConnectionHandler.cpp
+++ b/qpid/cpp/src/qpid/client/ConnectionHandler.cpp
@@ -21,7 +21,7 @@
 
 #include "qpid/client/ConnectionHandler.h"
 
-#include "qpid/client/SaslFactory.h"
+#include "qpid/SaslFactory.h"
 #include "qpid/client/Bounds.h"
 #include "qpid/framing/amqp_framing.h"
 #include "qpid/framing/all_method_bodies.h"
@@ -208,7 +208,13 @@ void ConnectionHandler::start(const FieldTable& /*serverProps*/, const Array& me
 {
     checkState(NOT_STARTED, INVALID_STATE_START);
     setState(NEGOTIATING);
-    sasl = SaslFactory::getInstance().create(*this);
+    sasl = SaslFactory::getInstance().create( username,
+                                              password,
+                                              service,
+                                              host,
+                                              minSsf,
+                                              maxSsf
+                                            );
 
     std::string mechlist;
     bool chosenMechanismSupported = mechanism.empty();
diff --git a/qpid/cpp/src/qpid/client/ConnectionHandler.h b/qpid/cpp/src/qpid/client/ConnectionHandler.h
index 61709db174..6af2e987fb 100644
--- a/qpid/cpp/src/qpid/client/ConnectionHandler.h
+++ b/qpid/cpp/src/qpid/client/ConnectionHandler.h
@@ -23,7 +23,7 @@
 
 #include "qpid/client/ChainableFrameHandler.h"
 #include "qpid/client/ConnectionSettings.h"
-#include "qpid/client/Sasl.h"
+#include "qpid/Sasl.h"
 #include "qpid/client/StateManager.h"
 #include "qpid/framing/AMQMethodBody.h"
 #include "qpid/framing/AMQP_HighestVersion.h"
diff --git a/qpid/cpp/src/tests/sasl.mk b/qpid/cpp/src/tests/sasl.mk
index 52cebe63f6..ae1666e891 100644
--- a/qpid/cpp/src/tests/sasl.mk
+++ b/qpid/cpp/src/tests/sasl.mk
@@ -26,6 +26,7 @@ cluster_authentication_soak_SOURCES=cluster_authentication_soak.cpp  ForkedBroke
 cluster_authentication_soak_LDADD=$(lib_client) $(lib_broker)
 
 TESTS +=   run_cluster_authentication_test
+TESTS +=   sasl_fed
 LONG_TESTS += run_cluster_authentication_soak
 
 endif # HAVE_SASL
diff --git a/qpid/cpp/src/tests/sasl_fed b/qpid/cpp/src/tests/sasl_fed
new file mode 100755
index 0000000000..550b5a1626
--- /dev/null
+++ b/qpid/cpp/src/tests/sasl_fed
@@ -0,0 +1,152 @@
+#! /bin/bash
+
+source test_env.sh
+
+minimum_sasl_version="2.1.22"
+if [ ! `pkg-config --atleast-version $minimum_sasl_version cyrus-sasl`]; then
+  echo "sasl_fed requires at least $minimum_sasl_version"
+  exit 0
+fi
+
+let minimum_sasl_version=$((2 * 65536 + 1 * 256 + 22))
+sasl_version_numbers=(`rpm -q cyrus-sasl-devel | head -1 | tr '-' ' ' | awk '{print $4}' | tr '.' ' '`)
+let sasl_version=$((${sasl_version_numbers[0]} * 65536 + ${sasl_version_numbers[1]} * 256 + ${sasl_version_numbers[2]}))
+
+if [ "$sasl_version" -lt "$minimum_sasl_version" ]; then
+  echo "sasl_fed requires version 2.1.22 or later"
+  exit 0
+fi
+
+exit
+
+QPID_SRC=$top_srcdir/src
+QPIDD=$QPID_SRC/.libs/qpidd
+PY_TOOLS=$QPID_TOOLS/src/py
+
+sasl_config_file=$QPID_SRC/tests/sasl_config
+
+my_random_number=$RANDOM
+tmp_root=/tmp/sasl_fed/$my_random_number
+mkdir -p $tmp_root
+
+
+#--------------------------------------------------
+#echo " Starting broker 1"
+#--------------------------------------------------
+$QPIDD                                       \
+  -p 0                                       \
+  --data-dir $tmp_root/data_1                \
+  --auth=yes                                 \
+  --mgmt-enable=yes                          \
+  --log-enable info+                         \
+  --log-source yes                           \
+  --log-to-file $tmp_root/qpidd_1.log        \
+  --sasl-config=$sasl_config_file            \
+  -d > $tmp_root/broker_1_port
+
+broker_1_port=`cat $tmp_root/broker_1_port`
+
+
+#--------------------------------------------------
+#echo " Starting broker 2"
+#--------------------------------------------------
+$QPIDD                                       \
+  -p 0                                       \
+  --data-dir $tmp_root/data_2                \
+  --auth=yes                                 \
+  --mgmt-enable=yes                          \
+  --log-enable info+                         \
+  --log-source yes                           \
+  --log-to-file $tmp_root/qpidd_2.log        \
+  --sasl-config=$sasl_config_file            \
+  -d > $tmp_root/broker_2_port
+
+broker_2_port=`cat $tmp_root/broker_2_port`
+
+
+# Now find the PIDs so I can kill them later.
+#pids=`ps -aef | grep -v grep | grep sasl_fed | grep $my_random_number  | awk '{print $2}'`
+
+
+# I am not randomizing these names, because the test creates its own brokers.
+QUEUE_NAME=sasl_fed_queue
+ROUTING_KEY=sasl_fed_queue
+EXCHANGE_NAME=sasl_fedex
+
+#--------------------------------------------------
+#echo "  add exchanges"
+#--------------------------------------------------
+$PY_TOOLS/qpid-config -a localhost:$broker_1_port add exchange direct $EXCHANGE_NAME
+$PY_TOOLS/qpid-config -a localhost:$broker_2_port add exchange direct $EXCHANGE_NAME
+
+
+#--------------------------------------------------
+#echo "  add queues"
+#--------------------------------------------------
+$PY_TOOLS/qpid-config -a localhost:$broker_1_port add queue $QUEUE_NAME
+$PY_TOOLS/qpid-config -a localhost:$broker_2_port add queue $QUEUE_NAME
+
+sleep 5
+
+#--------------------------------------------------
+#echo " create bindings"
+#--------------------------------------------------
+$PY_TOOLS/qpid-config -a localhost:$broker_1_port bind $EXCHANGE_NAME $QUEUE_NAME $ROUTING_KEY
+$PY_TOOLS/qpid-config -a localhost:$broker_2_port bind $EXCHANGE_NAME $QUEUE_NAME $ROUTING_KEY
+
+sleep 5
+
+
+#--------------------------------------------------
+#echo "  qpid-route route add"
+#--------------------------------------------------
+$PY_TOOLS/qpid-route route add zag/zag@localhost:$broker_2_port zag/zag@localhost:$broker_1_port $EXCHANGE_NAME $ROUTING_KEY "" "" DIGEST-MD5
+
+sleep 5
+
+
+n_messages=100
+#--------------------------------------------------
+#echo "  Sending 100 messages to $broker_1_port "
+#--------------------------------------------------
+$QPID_SRC/tests/datagen --count $n_messages | $QPID_SRC/tests/sender --username zag --password zag --exchange $EXCHANGE_NAME --routing-key $ROUTING_KEY --port $broker_1_port
+
+sleep 5
+
+#--------------------------------------------------
+#echo "  Examine Broker $broker_1_port"
+#--------------------------------------------------
+broker_1_message_count=`$PY_TOOLS/qpid-stat -q localhost:$broker_1_port | grep sasl_fed_queue | awk '{print $2}'`
+#echo " "
+
+#--------------------------------------------------
+#echo "  Examine Broker $broker_2_port"
+#--------------------------------------------------
+broker_2_message_count=`$PY_TOOLS/qpid-stat -q localhost:$broker_2_port | grep sasl_fed_queue | awk '{print $2}'`
+#echo " "
+
+#--------------------------------------------------
+#echo "  Asking brokers to quit."
+#--------------------------------------------------
+$QPIDD --port $broker_1_port --quit
+$QPIDD --port $broker_2_port --quit
+
+
+#--------------------------------------------------
+#echo   "Removing temporary directory $tmp_root"
+#--------------------------------------------------
+rm -rf $tmp_root
+
+if [ "$broker_2_message_count" =  "$n_messages" ]; then
+  echo "good: $broker_2_message_count"
+  exit 0
+else
+  echo "not ideal: $broker_1_message_count != $n_messages"
+  exit 1
+fi
+
+
+
+
+
+
diff --git a/qpid/tools/src/py/qpid-route b/qpid/tools/src/py/qpid-route
index be6bdf958c..0b5069a277 100755
--- a/qpid/tools/src/py/qpid-route
+++ b/qpid/tools/src/py/qpid-route
@@ -30,7 +30,7 @@ def Usage(short=False):
     print "Usage:  qpid-route [OPTIONS] dynamic add <dest-broker> <src-broker> <exchange> [tag] [exclude-list]"
     print "        qpid-route [OPTIONS] dynamic del <dest-broker> <src-broker> <exchange>"
     print
-    print "        qpid-route [OPTIONS] route add   <dest-broker> <src-broker> <exchange> <routing-key> [tag] [exclude-list]"
+    print "        qpid-route [OPTIONS] route add   <dest-broker> <src-broker> <exchange> <routing-key> [tag] [exclude-list] [mechanism]"
     print "        qpid-route [OPTIONS] route del   <dest-broker> <src-broker> <exchange> <routing-key>"
     print "        qpid-route [OPTIONS] queue add   <dest-broker> <src-broker> <exchange> <queue>"
     print "        qpid-route [OPTIONS] queue del   <dest-broker> <src-broker> <exchange> <queue>"
@@ -98,7 +98,7 @@ class RouteManager:
                 return link
         return None
 
-    def addLink(self, remoteBroker):
+    def addLink(self, remoteBroker, mech="PLAIN"):
         self.remote = BrokerURL(remoteBroker)
         if self.local.match(self.remote.host, self.remote.port):
             raise Exception("Linking broker to itself is not permitted")
@@ -107,10 +107,6 @@ class RouteManager:
         broker = brokers[0]
         link = self.getLink()
         if link == None:
-            if not self.remote.authName or self.remote.authName == "anonymous":
-                mech = "ANONYMOUS"
-            else:
-                mech = "PLAIN"
             res = broker.connect(self.remote.host, self.remote.port, _durable,
                                  mech, self.remote.authName or "", self.remote.authPass or "",
                                  _transport)
@@ -231,11 +227,11 @@ class RouteManager:
             if b[0] != self.local.name():
                 self.qmf.delBroker(b[1])
 
-    def addRoute(self, remoteBroker, exchange, routingKey, tag, excludes, dynamic=False):
+    def addRoute(self, remoteBroker, exchange, routingKey, tag, excludes, mech="PLAIN", dynamic=False):
         if dynamic and _srclocal:
             raise Exception("--src-local is not permitted on dynamic routes")
 
-        self.addLink(remoteBroker)
+        self.addLink(remoteBroker, mech)
         link = self.getLink()
         if link == None:
             raise Exception("Link failed to create")
@@ -494,9 +490,10 @@ try:
 
             tag = ""
             excludes = ""
+            mech = "PLAIN"
             if nargs > 5: tag = cargs[5]     
             if nargs > 6: excludes = cargs[6]     
-            rm.addRoute(remoteBroker, cargs[4], "", tag, excludes, dynamic=True)
+            rm.addRoute(remoteBroker, cargs[4], "", tag, excludes, mech, dynamic=True)
         elif cmd == "del":
             if nargs != 5:
                 Usage()
@@ -505,14 +502,16 @@ try:
 
     elif group == "route":
         if cmd == "add":
-            if nargs < 6 or nargs > 8:
+            if nargs < 6 or nargs > 9:
                 Usage()
 
             tag = ""
             excludes = ""
+            mech = "PLAIN"
             if nargs > 6: tag = cargs[6]     
             if nargs > 7: excludes = cargs[7]
-            rm.addRoute(remoteBroker, cargs[4], cargs[5], tag, excludes, dynamic=False)
+            if nargs > 8: mech = cargs[8]
+            rm.addRoute(remoteBroker, cargs[4], cargs[5], tag, excludes, mech, dynamic=False)
         elif cmd == "del":
             if nargs != 6:
                 Usage()