QPID-5280: prevent bind/unbind of someone elses exclusive queue

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

5 files changed, 33 insertions, 6 deletions

diff --git a/qpid/cpp/src/qpid/broker/Broker.cpp b/qpid/cpp/src/qpid/broker/Broker.cpp
index cf2912a9b1..da6c2d4a84 100644
--- a/qpid/cpp/src/qpid/broker/Broker.cpp
+++ b/qpid/cpp/src/qpid/broker/Broker.cpp
@@ -795,7 +795,7 @@ void Broker::createObject(const std::string& type, const std::string& name,
         framing::FieldTable arguments;
         qpid::amqp_0_10::translate(extensions, arguments);
 
-        bind(binding.queue, binding.exchange, binding.key, arguments, userId, connectionId);
+        bind(binding.queue, binding.exchange, binding.key, arguments, 0, userId, connectionId);
 
     } else if (type == TYPE_LINK) {
 
@@ -935,7 +935,7 @@ void Broker::deleteObject(const std::string& type, const std::string& name,
         deleteExchange(name, userId, connectionId);
     } else if (type == TYPE_BINDING) {
         BindingIdentifier binding(name);
-        unbind(binding.queue, binding.exchange, binding.key, userId, connectionId);
+        unbind(binding.queue, binding.exchange, binding.key, 0, userId, connectionId);
     } else if (type == TYPE_LINK) {
         boost::shared_ptr<Link> link = links.getLink(name);
         if (link) {
@@ -1432,6 +1432,7 @@ void Broker::bind(const std::string& queueName,
                   const std::string& exchangeName,
                   const std::string& key,
                   const qpid::framing::FieldTable& arguments,
+                  const OwnershipToken* owner,
                   const std::string& userId,
                   const std::string& connectionId)
 {
@@ -1453,6 +1454,9 @@ void Broker::bind(const std::string& queueName,
         throw framing::NotFoundException(QPID_MSG("Bind failed. No such queue: " << queueName));
     } else if (!exchange) {
         throw framing::NotFoundException(QPID_MSG("Bind failed. No such exchange: " << exchangeName));
+    } else if (queue->hasExclusiveOwner() && !queue->isExclusiveOwner(owner)) {
+        throw framing::ResourceLockedException(QPID_MSG("Cannot bind queue "
+                                               << queue->getName() << "; it is exclusive to another session"));
     } else {
         if (queue->bind(exchange, key, arguments)) {
             getBrokerObservers().bind(exchange, queue, key, arguments);
@@ -1473,6 +1477,7 @@ void Broker::bind(const std::string& queueName,
 void Broker::unbind(const std::string& queueName,
                     const std::string& exchangeName,
                     const std::string& key,
+                    const OwnershipToken* owner,
                     const std::string& userId,
                     const std::string& connectionId)
 {
@@ -1492,6 +1497,9 @@ void Broker::unbind(const std::string& queueName,
         throw framing::NotFoundException(QPID_MSG("Unbind failed. No such queue: " << queueName));
     } else if (!exchange) {
         throw framing::NotFoundException(QPID_MSG("Unbind failed. No such exchange: " << exchangeName));
+    } else if (queue->hasExclusiveOwner() && !queue->isExclusiveOwner(owner)) {
+        throw framing::ResourceLockedException(QPID_MSG("Cannot unbind queue "
+                                               << queue->getName() << "; it is exclusive to another session"));
     } else {
         if (exchange->unbind(queue, key, 0)) {
             if (exchange->isDurable() && queue->isDurable()) {
diff --git a/qpid/cpp/src/qpid/broker/Broker.h b/qpid/cpp/src/qpid/broker/Broker.h
index 3e22fb491b..e658d6f03b 100644
--- a/qpid/cpp/src/qpid/broker/Broker.h
+++ b/qpid/cpp/src/qpid/broker/Broker.h
@@ -342,6 +342,7 @@ class Broker : public sys::Runnable, public Plugin::Target,
         const std::string& exchange,
         const std::string& key,
         const qpid::framing::FieldTable& arguments,
+        const OwnershipToken* owner,
         const std::string& userId,
         const std::string& connectionId);
 
@@ -349,6 +350,7 @@ class Broker : public sys::Runnable, public Plugin::Target,
         const std::string& queue,
         const std::string& exchange,
         const std::string& key,
+        const OwnershipToken* owner,
         const std::string& userId,
         const std::string& connectionId);
 
diff --git a/qpid/cpp/src/qpid/broker/SemanticState.cpp b/qpid/cpp/src/qpid/broker/SemanticState.cpp
index 3e2a4cbb3c..ce7071c5ae 100644
--- a/qpid/cpp/src/qpid/broker/SemanticState.cpp
+++ b/qpid/cpp/src/qpid/broker/SemanticState.cpp
@@ -883,10 +883,10 @@ void SemanticState::unbindSessionBindings()
                 fedArguments.setString(qpidFedOp, fedOpUnbind);
                 fedArguments.setString(qpidFedOrigin, fedOrigin);
                 session.getBroker().bind(i->get<0>(), i->get<1>(), i->get<2>(), fedArguments,
-                                           userID, connectionId);
+                                         &session, userID, connectionId);
             } else {
                 session.getBroker().unbind(i->get<0>(), i->get<1>(), i->get<2>(),
-                                           userID, connectionId);
+                                           &session, userID, connectionId);
             }
         }
         catch (...) {
diff --git a/qpid/cpp/src/qpid/broker/SessionAdapter.cpp b/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
index 64e3fe1b8a..eff77db02f 100644
--- a/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
+++ b/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
@@ -155,7 +155,7 @@ void SessionAdapter::ExchangeHandlerImpl::bind(const string& queueName,
                                                const string& exchangeName, const string& routingKey,
                                                const FieldTable& arguments)
 {
-    getBroker().bind(queueName, exchangeName, routingKey, arguments,
+    getBroker().bind(queueName, exchangeName, routingKey, arguments, &session,
                      getConnection().getUserId(), getConnection().getMgmtId());
     state.addBinding(queueName, exchangeName, routingKey, arguments);
 }
@@ -165,7 +165,7 @@ void SessionAdapter::ExchangeHandlerImpl::unbind(const string& queueName,
                                                  const string& routingKey)
 {
     state.removeBinding(queueName, exchangeName, routingKey);
-    getBroker().unbind(queueName, exchangeName, routingKey,
+    getBroker().unbind(queueName, exchangeName, routingKey, &session,
                        getConnection().getUserId(), getConnection().getMgmtId());
 }
 
diff --git a/qpid/tests/src/py/qpid_tests/broker_0_10/queue.py b/qpid/tests/src/py/qpid_tests/broker_0_10/queue.py
index c62a1ae0c1..7e8218d092 100644
--- a/qpid/tests/src/py/qpid_tests/broker_0_10/queue.py
+++ b/qpid/tests/src/py/qpid_tests/broker_0_10/queue.py
@@ -92,6 +92,7 @@ class QueueTests(TestBase010):
 
         #declare an exclusive queue:
         s1.queue_declare(queue="exclusive-queue", exclusive=True, auto_delete=True)
+        s1.exchange_bind(exchange="amq.fanout", queue="exclusive-queue")
         try:
             #other connection should not be allowed to declare this:
             s2.queue_declare(queue="exclusive-queue", exclusive=True, auto_delete=True)
@@ -115,6 +116,22 @@ class QueueTests(TestBase010):
         except SessionException, e:
             self.assertEquals(405, e.args[0].error_code)
 
+        s5 = self.conn.session("binder")
+        try:
+            #other connection should not be allowed to declare this:
+            s5.exchange_bind(exchange="amq.direct", queue="exclusive-queue", binding_key="abc")
+            self.fail("Expected exchange_bind on an exclusive queue to raise an exception")
+        except SessionException, e:
+            self.assertEquals(405, e.args[0].error_code)
+
+        s6 = self.conn.session("unbinder")
+        try:
+            #other connection should not be allowed to declare this:
+            s6.exchange_unbind(exchange="amq.fanout", queue="exclusive-queue")
+            self.fail("Expected exchange_unbind on an exclusive queue to raise an exception")
+        except SessionException, e:
+            self.assertEquals(405, e.args[0].error_code)
+
     def test_declare_passive(self):
         """
         Test that the passive field is honoured in queue.declare