QPID-3652: Fix cluster authentication.

Only allow brokers that authenticate as the cluster-username to join a cluster.

New broker first connects to  a cluster broker authenticates as the cluster-username
and sends its CPG member ID to the qpid.cluster-credentials exchange.
The cluster broker that subsequently acts as updater verifies that the credentials are
valid before connecting to give the update.

NOTE 1: If you are using an ACL, the cluster-username must be allowed to
publish to the qpid.cluster-credentials exchange. E.g. in your ACL file:

acl allow foo@QPID publish exchange name=qpid.cluster-credentials

NOTE 2: This changes the cluster initialization protocol, you will
need to restart the cluster with all new version brokers.

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

7 files changed, 237 insertions, 8 deletions

diff --git a/cpp/src/qpid/cluster/Cluster.cpp b/cpp/src/qpid/cluster/Cluster.cpp
index 0241b0946b..40bfcd9285 100644
--- a/cpp/src/qpid/cluster/Cluster.cpp
+++ b/cpp/src/qpid/cluster/Cluster.cpp
@@ -130,6 +130,7 @@
 #include "qpid/cluster/UpdateDataExchange.h"
 #include "qpid/cluster/UpdateExchange.h"
 #include "qpid/cluster/ClusterTimer.h"
+#include "qpid/cluster/CredentialsExchange.h"
 
 #include "qpid/assert.h"
 #include "qmf/org/apache/qpid/cluster/ArgsClusterStopClusterNode.h"
@@ -162,6 +163,7 @@
 #include "qpid/framing/MessageTransferBody.h"
 #include "qpid/log/Helpers.h"
 #include "qpid/log/Statement.h"
+#include "qpid/UrlArray.h"
 #include "qpid/management/ManagementAgent.h"
 #include "qpid/memory.h"
 #include "qpid/sys/Thread.h"
@@ -189,6 +191,7 @@ using management::ManagementObject;
 using management::Manageable;
 using management::Args;
 namespace _qmf = ::qmf::org::apache::qpid::cluster;
+namespace arg=client::arg;
 
 /**
  * NOTE: must increment this number whenever any incompatible changes in
@@ -199,7 +202,7 @@ namespace _qmf = ::qmf::org::apache::qpid::cluster;
  * Currently use SVN revision to avoid clashes with versions from
  * different branches.
  */
-const uint32_t Cluster::CLUSTER_VERSION = 1159330;
+const uint32_t Cluster::CLUSTER_VERSION = 1207877;
 
 struct ClusterDispatcher : public framing::AMQP_AllOperations::ClusterHandler {
     qpid::cluster::Cluster& cluster;
@@ -211,12 +214,12 @@ struct ClusterDispatcher : public framing::AMQP_AllOperations::ClusterHandler {
 
     void initialStatus(uint32_t version, bool active, const Uuid& clusterId,
                        uint8_t storeState, const Uuid& shutdownId,
-                       const std::string& firstConfig)
+                       const std::string& firstConfig, const framing::Array& urls)
     {
         cluster.initialStatus(
             member, version, active, clusterId,
             framing::cluster::StoreState(storeState), shutdownId,
-            firstConfig, l);
+            firstConfig, urls, l);
     }
     void ready(const std::string& url) {
         cluster.ready(member, url, l);
@@ -267,6 +270,7 @@ Cluster::Cluster(const ClusterSettings& set, broker::Broker& b) :
                       poller),
     failoverExchange(new FailoverExchange(broker.GetVhostObject(), &broker)),
     updateDataExchange(new UpdateDataExchange(*this)),
+    credentialsExchange(new CredentialsExchange(*this)),
     quorum(boost::bind(&Cluster::leave, this)),
     decoder(boost::bind(&Cluster::deliverFrame, this, _1)),
     discarding(true),
@@ -300,6 +304,9 @@ Cluster::Cluster(const ClusterSettings& set, broker::Broker& b) :
     // for single control frame.
     broker.getExchanges().registerExchange(updateDataExchange);
 
+    // CredentialsExchange is used to authenticate new cluster members
+    broker.getExchanges().registerExchange(credentialsExchange);
+
     // Load my store status before we go into initialization
     if (! broker::NullMessageStore::isNullStore(&broker.getStore())) {
         store.load();
@@ -661,6 +668,7 @@ void Cluster::initMapCompleted(Lock& l) {
         setClusterId(initMap.getClusterId(), l);
 
         if (initMap.isUpdateNeeded())  { // Joining established cluster.
+            authenticate();
             broker.setRecovery(false); // Ditch my current store.
             broker.setClusterUpdatee(true);
             if (mAgent) mAgent->suppress(true); // Suppress mgmt output during update.
@@ -711,7 +719,8 @@ void Cluster::configChange(const MemberId&,
             ClusterInitialStatusBody(
                 ProtocolVersion(), CLUSTER_VERSION, state > INIT, clusterId,
                 store.getState(), store.getShutdownId(),
-                initMap.getFirstConfigStr()
+                initMap.getFirstConfigStr(),
+                vectorToUrlArray(getUrls(l))
             ),
             self);
     }
@@ -803,6 +812,7 @@ void Cluster::initialStatus(const MemberId& member, uint32_t version, bool activ
                             framing::cluster::StoreState store,
                             const framing::Uuid& shutdownId,
                             const std::string& firstConfig,
+                            const framing::Array& urls,
                             Lock& l)
 {
     if (version != CLUSTER_VERSION) {
@@ -816,7 +826,7 @@ void Cluster::initialStatus(const MemberId& member, uint32_t version, bool activ
     initMap.received(
         member,
         ClusterInitialStatusBody(ProtocolVersion(), version, active, id,
-                                 store, shutdownId, firstConfig)
+                                 store, shutdownId, firstConfig, urls)
     );
     if (initMap.transitionToComplete()) initMapCompleted(l);
 }
@@ -903,6 +913,11 @@ void Cluster::retractOffer(const MemberId& updater, uint64_t updateeInt, Lock& l
 }
 
 void Cluster::updateStart(const MemberId& updatee, const Url& url, Lock& l) {
+    // Check for credentials if authentication is enabled.
+    if (broker.getOptions().auth && !credentialsExchange->check(updatee)) {
+        QPID_LOG(error, "Un-authenticated attempt to join the cluster");
+        return;
+    }
     // NOTE: deliverEventQueue is already stopped at the stall point by deliveredEvent.
     if (state == LEFT) return;
     assert(state == OFFER);
@@ -1115,6 +1130,35 @@ void Cluster::updateMgmtMembership(Lock& l) {
     mgmtObject->set_memberIDs(idstr);
 }
 
+namespace {
+template <class T> struct AutoClose {
+    T closeme;
+    AutoClose(T t) : closeme(t) {}
+    ~AutoClose() { closeme.close(); }
+};
+}
+
+// Updatee connects to established member and stores credentials
+// in the qpid.cluster-credentials exchange to prove it
+// is safe for updater to connect and give an update.
+void Cluster::authenticate() {
+    if (!broker.getOptions().auth) return;
+    std::vector<Url> urls = initMap.getUrls();
+    for (std::vector<Url>::iterator i = urls.begin(); i != urls.end(); ++i) {
+        if (!i->empty()) {
+            client::Connection c;
+            c.open(*i, connectionSettings(settings));
+            AutoClose<client::Connection> closeConnection(c);
+            client::Session s = c.newSession(CredentialsExchange::NAME);
+            AutoClose<client::Session> closeSession(s);
+            client::Message credentials;
+            credentials.getHeaders().setUInt64(CredentialsExchange::NAME, getId());
+            s.messageTransfer(arg::content=credentials, arg::destination=CredentialsExchange::NAME);
+            s.sync();
+        }
+    }
+}
+
 std::ostream& operator<<(std::ostream& o, const Cluster& cluster) {
     static const char* STATE[] = {
         "PRE_INIT", "INIT", "JOINER", "UPDATEE", "CATCHUP",
diff --git a/cpp/src/qpid/cluster/Cluster.h b/cpp/src/qpid/cluster/Cluster.h
index ccec4948e6..f517c1b8d0 100644
--- a/cpp/src/qpid/cluster/Cluster.h
+++ b/cpp/src/qpid/cluster/Cluster.h
@@ -77,6 +77,7 @@ class Connection;
 struct EventFrame;
 class ClusterTimer;
 class UpdateDataExchange;
+class CredentialsExchange;
 
 /**
  * Connection to the cluster
@@ -187,6 +188,7 @@ class Cluster : private Cpg::Handler, public management::Manageable {
                        framing::cluster::StoreState,
                        const framing::Uuid& shutdownId,
                        const std::string& firstConfig,
+                       const framing::Array& urls,
                        Lock&);
     void ready(const MemberId&, const std::string&, Lock&);
     void configChange(const MemberId&,
@@ -215,6 +217,7 @@ class Cluster : private Cpg::Handler, public management::Manageable {
     void becomeElder(Lock&);
     void setMgmtStatus(Lock&);
     void updateMgmtMembership(Lock&);
+    void authenticate();
 
     // == Called in CPG dispatch thread
     void deliver( // CPG deliver callback.
@@ -271,6 +274,7 @@ class Cluster : private Cpg::Handler, public management::Manageable {
     PollableFrameQueue deliverFrameQueue;
     boost::shared_ptr<FailoverExchange> failoverExchange;
     boost::shared_ptr<UpdateDataExchange> updateDataExchange;
+    boost::shared_ptr<CredentialsExchange> credentialsExchange;
     Quorum quorum;
     LockedConnectionMap localConnections;
 
diff --git a/cpp/src/qpid/cluster/CredentialsExchange.cpp b/cpp/src/qpid/cluster/CredentialsExchange.cpp
new file mode 100644
index 0000000000..0fafc521cd
--- /dev/null
+++ b/cpp/src/qpid/cluster/CredentialsExchange.cpp
@@ -0,0 +1,94 @@
+/*
+ *
+ * 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 "CredentialsExchange.h"
+#include "Cluster.h"
+#include "qpid/broker/ConnectionState.h"
+#include "qpid/framing/reply_exceptions.h"
+#include "qpid/sys/Time.h"
+
+namespace qpid {
+namespace cluster {
+
+using namespace std;
+
+const string CredentialsExchange::NAME=("qpid.cluster-credentials");
+
+namespace {
+const string ANONYMOUS_MECH("ANONYMOUS");
+const string ANONYMOUS_USER("anonymous");
+
+string effectiveUserId(const string& username, const string& mechanism) {
+    if (mechanism == ANONYMOUS_MECH && username.empty())
+        return ANONYMOUS_USER;
+    else
+        return username;
+}
+}
+
+CredentialsExchange::CredentialsExchange(Cluster& cluster)
+    : broker::Exchange(NAME, &cluster),
+      username(effectiveUserId(cluster.getSettings().username,
+                               cluster.getSettings().mechanism)),
+      timeout(120*sys::TIME_SEC),
+      authenticate(cluster.getBroker().getOptions().auth)
+{}
+
+static const string anonymous("anonymous");
+
+bool CredentialsExchange::check(MemberId member) {
+    sys::Mutex::ScopedLock l(lock);
+    Map::iterator i = map.find(member);
+    if (i == map.end()) return false;
+    bool valid = (sys::Duration(i->second, sys::AbsTime::now()) < timeout);
+    map.erase(i);
+    return valid;
+}
+
+void CredentialsExchange::route(broker::Deliverable& msg, const string& /*routingKey*/, const framing::FieldTable* args) {
+    sys::Mutex::ScopedLock l(lock);
+    const broker::ConnectionState* connection =
+        static_cast<const broker::ConnectionState*>(msg.getMessage().getPublisher());
+    if (authenticate && !connection->isAuthenticatedUser(username))
+        throw framing::UnauthorizedAccessException(
+            QPID_MSG("Unauthorized user " << connection->getUserId() << " for " << NAME
+                     << ", should be " << username));
+    if (!args || !args->isSet(NAME))
+        throw framing::InvalidArgumentException(
+            QPID_MSG("Invalid message received by " << NAME));
+    MemberId member(args->getAsUInt64(NAME));
+    map[member] = sys::AbsTime::now();
+}
+
+string CredentialsExchange::getType() const { return NAME; }
+
+namespace {
+void throwIllegal() {
+    throw framing::NotAllowedException(
+        QPID_MSG("Illegal use of " << CredentialsExchange::NAME+" exchange"));
+}
+}
+
+bool CredentialsExchange::bind(boost::shared_ptr<broker::Queue> , const string& /*routingKey*/, const framing::FieldTable* ) { throwIllegal(); return false; }
+bool CredentialsExchange::unbind(boost::shared_ptr<broker::Queue> , const string& /*routingKey*/, const framing::FieldTable* ) { throwIllegal(); return false; }
+bool CredentialsExchange::isBound(boost::shared_ptr<broker::Queue>, const string* const /*routingKey*/, const framing::FieldTable* const ) { throwIllegal(); return false; }
+
+
+}} // Namespace qpid::cluster
diff --git a/cpp/src/qpid/cluster/CredentialsExchange.h b/cpp/src/qpid/cluster/CredentialsExchange.h
new file mode 100644
index 0000000000..90fd188271
--- /dev/null
+++ b/cpp/src/qpid/cluster/CredentialsExchange.h
@@ -0,0 +1,72 @@
+#ifndef QPID_CLUSTER_CREDENTIALSEXCHANGE_H
+#define QPID_CLUSTER_CREDENTIALSEXCHANGE_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 "types.h"
+#include <qpid/broker/Exchange.h>
+#include <qpid/sys/Mutex.h>
+#include <qpid/sys/Time.h>
+#include <string>
+#include <map>
+
+namespace qpid {
+namespace cluster {
+
+class Cluster;
+
+/**
+ * New members joining the cluster send their identity information to this
+ * exchange to prove they are authenticated as the cluster user.
+ * The exchange rejects messages that are not properly authenticated
+ */
+class CredentialsExchange : public broker::Exchange
+{
+  public:
+    static const std::string NAME;
+
+    CredentialsExchange(Cluster&);
+
+    /** Check if this member has credentials. The credentials are deleted. */
+    bool check(MemberId member);
+
+    /** Throw an exception if the calling connection is not the cluster user. Store credentials in msg. */
+    void route(broker::Deliverable& msg, const std::string& routingKey, const framing::FieldTable* args);
+
+    // Exchange overrides
+    std::string getType() const;
+    bool bind(boost::shared_ptr<broker::Queue> queue, const std::string& routingKey, const framing::FieldTable* args);
+    bool unbind(boost::shared_ptr<broker::Queue> queue, const std::string& routingKey, const framing::FieldTable* args);
+    bool isBound(boost::shared_ptr<broker::Queue> queue, const std::string* const routingKey, const framing::FieldTable* const args);
+
+  private:
+    typedef std::map<MemberId, sys::AbsTime> Map;
+    sys::Mutex lock;
+    Map map;
+    std::string username;
+    sys::Duration timeout;
+    bool authenticate;
+};
+
+}} // namespace qpid::cluster
+
+#endif  /*!QPID_CLUSTER_CREDENTIALSEXCHANGE_H*/
diff --git a/cpp/src/qpid/cluster/FailoverExchange.cpp b/cpp/src/qpid/cluster/FailoverExchange.cpp
index cfbe34a460..43ec27cf2c 100644
--- a/cpp/src/qpid/cluster/FailoverExchange.cpp
+++ b/cpp/src/qpid/cluster/FailoverExchange.cpp
@@ -28,6 +28,7 @@
 #include "qpid/framing/MessageTransferBody.h"
 #include "qpid/log/Statement.h"
 #include "qpid/framing/Array.h"
+#include "qpid/UrlArray.h"
 #include <boost/bind.hpp>
 #include <algorithm>
 
@@ -86,9 +87,7 @@ void FailoverExchange::route(Deliverable&, const string& , const framing::FieldT
 void FailoverExchange::sendUpdate(const Queue::shared_ptr& queue) {
     // Called with lock held.
     if (urls.empty()) return;
-    framing::Array array(0x95);
-    for (Urls::const_iterator i = urls.begin(); i != urls.end(); ++i)
-        array.add(boost::shared_ptr<Str16Value>(new Str16Value(i->str())));
+    framing::Array array = vectorToUrlArray(urls);
     const ProtocolVersion v;
     boost::intrusive_ptr<Message> msg(new Message);
     AMQFrame command(MessageTransferBody(v, typeName, 1, 0));
diff --git a/cpp/src/qpid/cluster/InitialStatusMap.cpp b/cpp/src/qpid/cluster/InitialStatusMap.cpp
index c8ecc13f2c..eb65005a9e 100644
--- a/cpp/src/qpid/cluster/InitialStatusMap.cpp
+++ b/cpp/src/qpid/cluster/InitialStatusMap.cpp
@@ -21,6 +21,7 @@
 #include "InitialStatusMap.h"
 #include "StoreStatus.h"
 #include "qpid/log/Statement.h"
+#include "qpid/UrlArray.h"
 #include <algorithm>
 #include <vector>
 #include <boost/bind.hpp>
@@ -218,6 +219,17 @@ void InitialStatusMap::checkConsistent() {
     }
 }
 
+std::vector<Url> InitialStatusMap::getUrls() const {
+    std::vector<Url> urls;
+    for (Map::const_iterator i = map.begin(); i != map.end(); ++i) {
+        if (i->second) {
+            std::vector<Url> urls = urlArrayToVector(i->second->getUrls());
+            if (!urls.empty()) return urls;
+        }
+    }
+    return std::vector<Url>();
+}
+
 std::string InitialStatusMap::getFirstConfigStr() const {
     assert(!firstConfig.empty());
     return encodeMemberSet(firstConfig);
diff --git a/cpp/src/qpid/cluster/InitialStatusMap.h b/cpp/src/qpid/cluster/InitialStatusMap.h
index a5a600365e..afa0110836 100644
--- a/cpp/src/qpid/cluster/InitialStatusMap.h
+++ b/cpp/src/qpid/cluster/InitialStatusMap.h
@@ -23,8 +23,10 @@
  */
 
 #include "MemberSet.h"
+#include "qpid/Url.h"
 #include <qpid/framing/ClusterInitialStatusBody.h>
 #include <boost/optional.hpp>
+#include <vector>
 
 namespace qpid {
 namespace cluster {
@@ -69,6 +71,8 @@ class InitialStatusMap
     framing::Uuid getClusterId();
     /**@pre isComplete(). @throw Exception if there are any inconsistencies. */
     void checkConsistent();
+    /*@return cluster URLs */
+    std::vector<Url> getUrls() const;
 
     /** Get first config-change for this member, encoded as a string.
      *@pre configChange has been called at least once.