QPID-3544: ACL denials while replicating exclusive queues to a newly joined node.

Disabled ACL during cluster update process.

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

4 files changed, 16 insertions, 7 deletions

diff --git a/cpp/src/qpid/acl/AclPlugin.cpp b/cpp/src/qpid/acl/AclPlugin.cpp
index e4d721ea44..d611797c49 100644
--- a/cpp/src/qpid/acl/AclPlugin.cpp
+++ b/cpp/src/qpid/acl/AclPlugin.cpp
@@ -69,7 +69,7 @@ struct AclPlugin : public Plugin {
     	}
 
         acl = new Acl(values, b);
-		b.setAcl(acl.get());
+        b.setAcl(acl.get());
         b.addFinalizer(boost::bind(&AclPlugin::shutdown, this));
     }
 
diff --git a/cpp/src/qpid/cluster/Cluster.cpp b/cpp/src/qpid/cluster/Cluster.cpp
index 7432fbbc33..e6e3de64f2 100644
--- a/cpp/src/qpid/cluster/Cluster.cpp
+++ b/cpp/src/qpid/cluster/Cluster.cpp
@@ -278,7 +278,8 @@ Cluster::Cluster(const ClusterSettings& set, broker::Broker& b) :
     lastBroker(false),
     updateRetracted(false),
     updateClosed(false),
-    error(*this)
+    error(*this),
+    acl(0)
 {
     broker.setInCluster(true);
 
@@ -856,6 +857,8 @@ void Cluster::updateOffer(const MemberId& updater, uint64_t updateeInt, Lock& l)
     else if (updatee == self && url) {
         assert(state == JOINER);
         state = UPDATEE;
+        acl = broker.getAcl();
+        broker.setAcl(0);       // Disable ACL during update
         QPID_LOG(notice, *this << " receiving update from " << updater);
         checkUpdateIn(l);
     }
@@ -956,6 +959,7 @@ void Cluster::checkUpdateIn(Lock& l) {
         // NB: don't updateMgmtMembership() here as we are not in the deliver
         // thread. It will be updated on delivery of the "ready" we just mcast.
         broker.setClusterUpdatee(false);
+        broker.setAcl(acl);     // Restore ACL
         discarding = false;     // OK to set, we're stalled for update.
         QPID_LOG(notice, *this << " update complete, starting catch-up.");
         QPID_LOG(debug, debugSnapshot()); // OK to call because we're stalled.
diff --git a/cpp/src/qpid/cluster/Cluster.h b/cpp/src/qpid/cluster/Cluster.h
index da5781b7a9..ccec4948e6 100644
--- a/cpp/src/qpid/cluster/Cluster.h
+++ b/cpp/src/qpid/cluster/Cluster.h
@@ -56,6 +56,7 @@ namespace qpid {
 
 namespace broker {
 class Message;
+class AclModule;
 }
 
 namespace framing {
@@ -312,6 +313,7 @@ class Cluster : private Cpg::Handler, public management::Manageable {
     sys::Timer clockTimer;
     sys::AbsTime clusterTime;
     sys::Duration clusterTimeOffset;
+    broker::AclModule* acl;
 
   friend std::ostream& operator<<(std::ostream&, const Cluster&);
   friend struct ClusterDispatcher;
diff --git a/cpp/src/tests/cluster_tests.py b/cpp/src/tests/cluster_tests.py
index 610f6f9a04..0e80e06d34 100755
--- a/cpp/src/tests/cluster_tests.py
+++ b/cpp/src/tests/cluster_tests.py
@@ -115,19 +115,22 @@ class ShortTests(BrokerTest):
         acl=os.path.join(os.getcwd(), "policy.acl")
         aclf=file(acl,"w")
         aclf.write("""
-acl deny zag@QPID create queue
-acl allow all all
+acl allow zig@QPID all all
+acl deny all all
 """)
         aclf.close()
-        cluster = self.cluster(2, args=["--auth", "yes",
+        cluster = self.cluster(1, args=["--auth", "yes",
                                         "--sasl-config", sasl_config,
                                         "--load-module", os.getenv("ACL_LIB"),
                                         "--acl-file", acl])
 
         # Valid user/password, ensure queue is created.
         c = cluster[0].connect(username="zig", password="zig")
-        c.session().sender("ziggy;{create:always}")
+        c.session().sender("ziggy;{create:always,node:{x-declare:{exclusive:true}}}")
         c.close()
+        cluster.start()                 # Start second node.
+
+        # Check queue is created on second node.
         c = cluster[1].connect(username="zig", password="zig")
         c.session().receiver("ziggy;{assert:always}")
         c.close()
@@ -156,7 +159,7 @@ acl allow all all
             self.fail("Expected exception")
         except messaging.exceptions.UnauthorizedAccess: pass
         # make sure the queue was not created at the other node.
-        c = cluster[0].connect(username="zag", password="zag")
+        c = cluster[1].connect(username="zig", password="zig")
         try:
             s = c.session()
             s.sender("zaggy;{assert:always}")