QPID-2393 Limit number of queues per user.

Merge work from branches/qpid-2393
This scheme works for old HA as long as cluster members run with the same --max-queues-per-user setting.



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

12 files changed, 366 insertions, 7 deletions

diff --git a/cpp/src/CMakeLists.txt b/cpp/src/CMakeLists.txt
index 681f095b09..6d32a79e63 100644
--- a/cpp/src/CMakeLists.txt
+++ b/cpp/src/CMakeLists.txt
@@ -600,6 +600,8 @@ if (BUILD_ACL)
        qpid/acl/AclPlugin.cpp
        qpid/acl/AclReader.cpp
        qpid/acl/AclReader.h
+       qpid/acl/AclResourceCounter.cpp
+       qpid/acl/AclResourceCounter.h
        qpid/acl/AclValidator.cpp
        qpid/acl/AclValidator.h
       )
diff --git a/cpp/src/acl.mk b/cpp/src/acl.mk
index 0301f8c754..74f8ef14ed 100644
--- a/cpp/src/acl.mk
+++ b/cpp/src/acl.mk
@@ -31,6 +31,8 @@ acl_la_SOURCES = \
   qpid/acl/AclPlugin.cpp \
   qpid/acl/AclReader.cpp \
   qpid/acl/AclReader.h \
+  qpid/acl/AclResourceCounter.cpp \
+  qpid/acl/AclResourceCounter.h \
   qpid/acl/AclValidator.cpp \
   qpid/acl/AclValidator.h
 
diff --git a/cpp/src/qpid/acl/Acl.cpp b/cpp/src/qpid/acl/Acl.cpp
index 89c4b3402a..6eeda9f26c 100644
--- a/cpp/src/qpid/acl/Acl.cpp
+++ b/cpp/src/qpid/acl/Acl.cpp
@@ -18,6 +18,7 @@
 
 #include "qpid/acl/Acl.h"
 #include "qpid/acl/AclConnectionCounter.h"
+#include "qpid/acl/AclResourceCounter.h"
 #include "qpid/acl/AclData.h"
 #include "qpid/acl/AclValidator.h"
 #include "qpid/sys/Mutex.h"
@@ -32,6 +33,7 @@
 #include "qmf/org/apache/qpid/acl/Package.h"
 #include "qmf/org/apache/qpid/acl/EventAllow.h"
 #include "qmf/org/apache/qpid/acl/EventConnectionDeny.h"
+#include "qmf/org/apache/qpid/acl/EventQueueQuotaDeny.h"
 #include "qmf/org/apache/qpid/acl/EventDeny.h"
 #include "qmf/org/apache/qpid/acl/EventFileLoaded.h"
 #include "qmf/org/apache/qpid/acl/EventFileLoadFailed.h"
@@ -51,8 +53,8 @@ using qpid::management::Args;
 namespace _qmf = qmf::org::apache::qpid::acl;
 
 Acl::Acl (AclValues& av, Broker& b): aclValues(av), broker(&b), transferAcl(false), mgmtObject(0),
-    connectionCounter(new ConnectionCounter(*this, aclValues.aclMaxConnectPerUser, aclValues.aclMaxConnectPerIp, aclValues.aclMaxConnectTotal))
-{
+    connectionCounter(new ConnectionCounter(*this, aclValues.aclMaxConnectPerUser, aclValues.aclMaxConnectPerIp, aclValues.aclMaxConnectTotal)),
+    resourceCounter(new ResourceCounter(*this, aclValues.aclMaxQueuesPerUser)){
 
     agent = broker->getManagementAgent();
 
@@ -63,6 +65,7 @@ Acl::Acl (AclValues& av, Broker& b): aclValues(av), broker(&b), transferAcl(fals
         mgmtObject->set_maxConnections(aclValues.aclMaxConnectTotal);
         mgmtObject->set_maxConnectionsPerIp(aclValues.aclMaxConnectPerIp);
         mgmtObject->set_maxConnectionsPerUser(aclValues.aclMaxConnectPerUser);
+        mgmtObject->set_maxQueuesPerUser(aclValues.aclMaxQueuesPerUser);
     }
     std::string errorString;
     if (!readAclFile(errorString)){
@@ -84,6 +87,15 @@ void Acl::reportConnectLimit(const std::string user, const std::string addr)
 }
 
 
+void Acl::reportQueueLimit(const std::string user, const std::string queueName)
+{
+    if (mgmtObject!=0)
+        mgmtObject->inc_queueQuotaDenyCount();
+
+    agent->raiseEvent(_qmf::EventQueueQuotaDeny(user, queueName));
+}
+
+
 bool Acl::authorise(
     const std::string&               id,
     const Action&                    action,
@@ -136,6 +148,18 @@ void Acl::setUserId(const qpid::broker::Connection& connection, const std::strin
 }
 
 
+bool Acl::approveCreateQueue(const std::string& userId, const std::string& queueName)
+{
+    return resourceCounter->approveCreateQueue(userId, queueName);
+}
+
+
+void Acl::recordDestroyQueue(const std::string& queueName)
+{
+    resourceCounter->recordDestroyQueue(queueName);
+}
+
+
 bool Acl::result(
     const AclResult&   aclreslt,
     const std::string& id,
diff --git a/cpp/src/qpid/acl/Acl.h b/cpp/src/qpid/acl/Acl.h
index 4787934275..e0513d5d1e 100644
--- a/cpp/src/qpid/acl/Acl.h
+++ b/cpp/src/qpid/acl/Acl.h
@@ -43,12 +43,14 @@ class Connection;
 
 namespace acl {
 class ConnectionCounter;
+class ResourceCounter;
 
 struct AclValues {
     std::string aclFile;
     uint16_t    aclMaxConnectPerUser;
     uint16_t    aclMaxConnectPerIp;
     uint16_t    aclMaxConnectTotal;
+    uint16_t    aclMaxQueuesPerUser;
 };
 
 
@@ -64,6 +66,7 @@ private:
     qpid::management::ManagementAgent*   agent;
     mutable qpid::sys::Mutex             dataLock;
     boost::shared_ptr<ConnectionCounter> connectionCounter;
+    boost::shared_ptr<ResourceCounter>   resourceCounter;
 
 public:
     Acl (AclValues& av, broker::Broker& b);
@@ -72,6 +75,7 @@ public:
      * issue management counts and alerts for denied connections
      */
     void reportConnectLimit(const std::string user, const std::string addr);
+    void reportQueueLimit(const std::string user, const std::string queueName);
 
     inline virtual bool doTransferAcl() {
         return transferAcl;
@@ -92,9 +96,11 @@ public:
         const std::string&               ExchangeName,
         const std::string&               RoutingKey);
 
+    // Resource quota tracking
     virtual bool approveConnection(const broker::Connection& connection);
-
     virtual void setUserId(const broker::Connection& connection, const std::string& username);
+    virtual bool approveCreateQueue(const std::string& userId, const std::string& queueName);
+    virtual void recordDestroyQueue(const std::string& queueName);
 
     virtual ~Acl();
 private:
diff --git a/cpp/src/qpid/acl/AclPlugin.cpp b/cpp/src/qpid/acl/AclPlugin.cpp
index ebf5e90afe..4aaa00a8b0 100644
--- a/cpp/src/qpid/acl/AclPlugin.cpp
+++ b/cpp/src/qpid/acl/AclPlugin.cpp
@@ -45,6 +45,7 @@ struct AclOptions : public Options {
             ("max-connections"         , optValue(values.aclMaxConnectTotal, "N"),   "The maximum combined number of connections allowed. 0 implies no limit.")
             ("max-connections-per-user", optValue(values.aclMaxConnectPerUser, "N"), "The maximum number of connections allowed per user. 0 implies no limit.")
             ("max-connections-per-ip"  , optValue(values.aclMaxConnectPerIp, "N"),   "The maximum number of connections allowed per host IP address. 0 implies no limit.")
+            ("max-queues-per-user",      optValue(values.aclMaxQueuesPerUser, "N"),  "The maximum number of queues allowed per user. 0 implies no limit.")
             ;
     }
 };
diff --git a/cpp/src/qpid/acl/AclResourceCounter.cpp b/cpp/src/qpid/acl/AclResourceCounter.cpp
new file mode 100644
index 0000000000..66dfd0777e
--- /dev/null
+++ b/cpp/src/qpid/acl/AclResourceCounter.cpp
@@ -0,0 +1,165 @@
+/*
+ *
+ * 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 "AclResourceCounter.h"
+#include "Acl.h"
+#include "qpid/log/Statement.h"
+#include "qpid/sys/Mutex.h"
+#include <assert.h>
+#include <sstream>
+
+using namespace qpid::sys;
+
+namespace qpid {
+namespace acl {
+
+//
+// This module approves various resource creation requests:
+//   Queues
+//
+
+
+//
+//
+//
+ResourceCounter::ResourceCounter(Acl& a, uint16_t ql) :
+    acl(a), queueLimit(ql) {}
+
+ResourceCounter::~ResourceCounter() {}
+
+
+//
+// limitApproveLH
+//
+// Resource creation approver.
+// If user is under limit increment count and return true.
+// Called with lock held.
+//
+bool ResourceCounter::limitApproveLH(
+    const std::string& theTitle,
+    countsMap_t& theMap,
+    const std::string& theName,
+    uint16_t theLimit,
+    bool emitLog) {
+
+    bool result(true);
+    if (theLimit > 0) {
+        uint16_t count;
+        countsMap_t::iterator eRef = theMap.find(theName);
+        if (eRef != theMap.end()) {
+            count = (uint16_t)(*eRef).second;
+            result = count < theLimit;
+            if (result) {
+                count += 1;
+                (*eRef).second = count;
+            }
+        } else {
+            // Not found
+            theMap[theName] = count = 1;
+        }
+        if (emitLog) {
+            QPID_LOG(trace, theTitle << theName
+                << " limit=" << theLimit
+                << " curValue=" << count
+                << " result=" << (result ? "allow" : "deny"));
+        }
+    }
+    return result;
+}
+
+
+//
+// releaseLH
+//
+// Decrement the name's count in map.
+// called with dataLock already taken
+//
+void ResourceCounter::releaseLH(
+    const std::string& theTitle, countsMap_t& theMap, const std::string& theName, uint16_t theLimit) {
+
+    if (theLimit > 0) {
+        countsMap_t::iterator eRef = theMap.find(theName);
+        if (eRef != theMap.end()) {
+            uint16_t count = (uint16_t) (*eRef).second;
+            assert (count > 0);
+            if (1 == count) {
+                theMap.erase (eRef);
+            } else {
+                (*eRef).second = count - 1;
+            }
+        } else {
+            // User had no connections.
+            QPID_LOG(notice, theTitle << theName
+                << "' not found in resource count pool");
+        }
+    }
+}
+
+
+//
+// approveCreateQueue
+//  Count an attempted queue creation by this user.
+//  Disapprove if over limit.
+//
+bool ResourceCounter::approveCreateQueue(const std::string& userId, const std::string& queueName)
+{
+    Mutex::ScopedLock locker(dataLock);
+
+    bool okByQ = limitApproveLH("ACL Queue creation approver. userId:", queuePerUserMap, userId, queueLimit, true);
+
+    if (okByQ) {
+        // Queue is owned by this userId
+        queueOwnerMap[queueName] = userId;
+
+        QPID_LOG(trace, "ACL create queue approved for user '" << userId
+            << "' queue '" << queueName << "'");
+    } else {
+
+        QPID_LOG(error, "Client max queue count limit of " << queueLimit
+            << " exceeded by '" << userId << "' creating queue '"
+            << queueName << "'. Queue creation denied.");
+
+        acl.reportQueueLimit(userId, queueName);
+    }
+    return okByQ;
+}
+
+
+//
+// recordDestroyQueue
+//  Return a destroyed queue to a user's quota
+//
+void ResourceCounter::recordDestroyQueue(const std::string& queueName)
+{
+    Mutex::ScopedLock locker(dataLock);
+
+    queueOwnerMap_t::iterator eRef = queueOwnerMap.find(queueName);
+    if (eRef != queueOwnerMap.end()) {
+        releaseLH("ACL resource counter: Queue owner for queue '", queuePerUserMap, (*eRef).second, queueLimit);
+
+        queueOwnerMap.erase(eRef);
+    } else {
+        QPID_LOG(notice, "ACL resource counter: Queue '" << queueName
+            << "' not found in queue owner map");
+    }
+}
+
+}} // namespace qpid::acl
diff --git a/cpp/src/qpid/acl/AclResourceCounter.h b/cpp/src/qpid/acl/AclResourceCounter.h
new file mode 100644
index 0000000000..f5995eb961
--- /dev/null
+++ b/cpp/src/qpid/acl/AclResourceCounter.h
@@ -0,0 +1,78 @@
+#ifndef QPID_ACL_RESOURCECOUNTER_H
+#define QPID_ACL_RESOURCECOUNTER_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 "qpid/sys/Mutex.h"
+#include <boost/iterator/iterator_concepts.hpp>
+
+#include <map>
+
+namespace qpid {
+
+namespace acl {
+class Acl;
+
+ /**
+ * Approve or disapprove resource creation requests
+ */
+class ResourceCounter
+{
+private:
+    typedef std::map<std::string, uint32_t> countsMap_t;
+    typedef std::map<std::string, std::string> queueOwnerMap_t;
+
+    Acl&             acl;
+    uint16_t         queueLimit;
+    qpid::sys::Mutex dataLock;
+
+    /** Records queueName-queueUserId */
+    queueOwnerMap_t queueOwnerMap;
+
+    /** Records queue-by-owner counts */
+    countsMap_t queuePerUserMap;
+
+    /** Return approval for proposed resource creation */
+    bool limitApproveLH(const std::string& theTitle,
+                        countsMap_t& theMap,
+                        const std::string& theName,
+                        uint16_t theLimit,
+                        bool emitLog);
+
+    /** Release a connection */
+    void releaseLH(const std::string& theTitle,
+                   countsMap_t& theMap,
+                   const std::string& theName,
+                   uint16_t theLimit);
+
+public:
+    ResourceCounter(Acl& acl, uint16_t ql);
+    ~ResourceCounter();
+
+    // Queue counting
+    bool approveCreateQueue(const std::string& userId, const std::string& queueName);
+    void recordDestroyQueue(const std::string& queueName);
+};
+
+}} // namespace qpid::acl
+
+#endif  /*!QPID_ACL_RESOURCECOUNTER_H*/
diff --git a/cpp/src/qpid/acl/management-schema.xml b/cpp/src/qpid/acl/management-schema.xml
index f52c251bed..5e42e94f52 100644
--- a/cpp/src/qpid/acl/management-schema.xml
+++ b/cpp/src/qpid/acl/management-schema.xml
@@ -25,8 +25,10 @@
     <property name="maxConnections"        type="uint16"  access="RO"       desc="Maximum allowed connections"/>
     <property name="maxConnectionsPerIp"   type="uint16"  access="RO"       desc="Maximum allowed connections"/>
     <property name="maxConnectionsPerUser" type="uint16"  access="RO"       desc="Maximum allowed connections"/>
+    <property name="maxQueuesPerUser"      type="uint16"  access="RO"       desc="Maximum allowed queues"/>
     <statistic name="aclDenyCount"         type="count64" unit="request"    desc="Number of ACL requests denied"/>
     <statistic name="connectionDenyCount"  type="count64" unit="connection" desc="Number of connections denied"/>
+    <statistic name="queueQuotaDenyCount"  type="count64" unit="connection" desc="Number of queue creations denied"/>
 
     <method name="reloadACLFile" desc="Reload the ACL file"/>
 
@@ -70,11 +72,13 @@
     <arg name="reason"     type="lstr"/>
     <arg name="userId"     type="sstr"/>
     <arg name="clientAddr" type="sstr"/>
+    <arg name="queueName"  type="sstr"/>
   </eventArguments>
 
   <event name="allow"          sev="inform" args="userId, action, objectType, objectName, arguments"/>
   <event name="deny"           sev="notice" args="userId, action, objectType, objectName, arguments"/>
   <event name="connectionDeny" sev="notice" args="userId, clientAddr"/>
+  <event name="queueQuotaDeny" sev="notice" args="userId, queueName"/>
   <event name="fileLoaded"     sev="inform" args="userId"/>
   <event name="fileLoadFailed" sev="error"  args="userId, reason"/>
 
diff --git a/cpp/src/qpid/broker/AclModule.h b/cpp/src/qpid/broker/AclModule.h
index 4caf8ed3ce..987d3e4a78 100644
--- a/cpp/src/qpid/broker/AclModule.h
+++ b/cpp/src/qpid/broker/AclModule.h
@@ -151,6 +151,11 @@ namespace broker {
          */
         virtual void setUserId(const Connection& connection, const std::string& username)=0;
 
+        /** Approve queue creation by counting per-user.
+         */
+        virtual bool approveCreateQueue(const std::string& userId, const std::string& queueName)=0;
+        virtual void recordDestroyQueue(const std::string& queueName)=0;
+
         virtual ~AclModule() {};
     };
 } // namespace broker
diff --git a/cpp/src/qpid/broker/Broker.cpp b/cpp/src/qpid/broker/Broker.cpp
index 02a6886f90..03ad88a4a8 100644
--- a/cpp/src/qpid/broker/Broker.cpp
+++ b/cpp/src/qpid/broker/Broker.cpp
@@ -1072,6 +1072,9 @@ std::pair<boost::shared_ptr<Queue>, bool> Broker::createQueue(
 
         if (!acl->authorise(userId,acl::ACT_CREATE,acl::OBJ_QUEUE,name,&params) )
             throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied queue create request from " << userId));
+
+        if (!acl->approveCreateQueue(userId,name) )
+            throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied queue create request from " << userId));
     }
 
     Exchange::shared_ptr alternate;
@@ -1116,6 +1119,8 @@ void Broker::deleteQueue(const std::string& name, const std::string& userId,
     Queue::shared_ptr queue = queues.find(name);
     if (queue) {
         if (check) check(queue);
+        if (acl)
+            acl->recordDestroyQueue(name);
         queues.destroy(name);
         queue->destroyed();
     } else {
diff --git a/cpp/src/tests/acl.py b/cpp/src/tests/acl.py
index 102796cba6..180f848104 100755
--- a/cpp/src/tests/acl.py
+++ b/cpp/src/tests/acl.py
@@ -53,6 +53,9 @@ class ACLTests(TestBase010):
     def port_u(self):
         return int(self.defines["port-u"])
 
+    def port_q(self):
+        return int(self.defines["port-q"])
+
     def get_session_by_port(self, user, passwd, byPort):
         socket = connect(self.broker.host, byPort)
         connection = Connection (sock=socket, username=user, password=passwd,
@@ -2243,6 +2246,62 @@ class ACLTests(TestBase010):
         self.LookupPublish("joe@QPID", "QPID-work",  "QPID",        "allow")
         self.LookupPublish("joe@QPID", "QPID-work2", "QPID",        "allow")
 
+   #=====================================
+   # Queue per-user quota
+   #=====================================
+
+    def test_queue_per_user_quota(self):
+        """
+        Test ACL queue counting limits.
+        port_q has a limit of 2
+        """
+        # bob should be able to create two queues
+        session = self.get_session_by_port('bob','bob', self.port_q())
+
+        try:
+            session.queue_declare(queue="queue1")
+            session.queue_declare(queue="queue2")
+        except qpid.session.SessionException, e:
+            self.fail("Error during queue create request");
+
+        # third queue should fail
+        try:
+            session.queue_declare(queue="queue3")
+            self.fail("Should not be able to create third queue")
+        except Exception, e:
+            result = None
+            session = self.get_session_by_port('bob','bob', self.port_q())
+
+        # alice should be able to create two queues
+        session2 = self.get_session_by_port('alice','alice', self.port_q())
+
+        try:
+            session2.queue_declare(queue="queuea1")
+            session2.queue_declare(queue="queuea2")
+        except qpid.session.SessionException, e:
+            self.fail("Error during queue create request");
+
+        # third queue should fail
+        try:
+            session2.queue_declare(queue="queuea3")
+            self.fail("Should not be able to create third queue")
+        except Exception, e:
+            result = None
+            session2 = self.get_session_by_port('alice','alice', self.port_q())
+
+        # bob should be able to delete a queue and create another
+        try:
+            session.queue_delete(queue="queue1")
+            session.queue_declare(queue="queue3")
+        except qpid.session.SessionException, e:
+            self.fail("Error during queue create request");
+
+        # alice should be able to delete a queue and create another
+        try:
+            session2.queue_delete(queue="queuea1")
+            session2.queue_declare(queue="queuea3")
+        except qpid.session.SessionException, e:
+            self.fail("Error during queue create request");
 
 class BrokerAdmin:
     def __init__(self, broker, username=None, password=None):
diff --git a/cpp/src/tests/run_acl_tests b/cpp/src/tests/run_acl_tests
index 25241ad75e..d4d0d74e00 100755
--- a/cpp/src/tests/run_acl_tests
+++ b/cpp/src/tests/run_acl_tests
@@ -24,22 +24,26 @@ source ./test_env.sh
 DATA_DIR=`pwd`/data_dir
 DATA_DIRI=`pwd`/data_diri
 DATA_DIRU=`pwd`/data_diru
+DATA_DIRQ=`pwd`/data_dirq
 
 trap stop_brokers INT TERM QUIT
 
 start_brokers() {
-    ../qpidd --daemon --port 0 --no-module-dir --data-dir $DATA_DIR --load-module $ACL_LIB --acl-file policy.acl --auth no --log-to-file local.log > qpidd.port
+    ../qpidd --daemon --port 0 --no-module-dir --data-dir $DATA_DIR --load-module $ACL_LIB --acl-file policy.acl --auth no                               --log-to-file local.log > qpidd.port
     LOCAL_PORT=`cat qpidd.port`
-    ../qpidd --daemon --port 0 --no-module-dir --data-dir $DATA_DIRI --load-module $ACL_LIB --acl-file policy.acl --auth no --max-connections-per-ip 2 --log-to-file locali.log > qpiddi.port
+    ../qpidd --daemon --port 0 --no-module-dir --data-dir $DATA_DIRI --load-module $ACL_LIB --acl-file policy.acl --auth no --max-connections-per-ip 2   --log-to-file locali.log > qpiddi.port
     LOCAL_PORTI=`cat qpiddi.port`
     ../qpidd --daemon --port 0 --no-module-dir --data-dir $DATA_DIRU --load-module $ACL_LIB --acl-file policy.acl --auth no --max-connections-per-user 2 --log-to-file localu.log > qpiddu.port
     LOCAL_PORTU=`cat qpiddu.port`
+    ../qpidd --daemon --port 0 --no-module-dir --data-dir $DATA_DIRQ --load-module $ACL_LIB --acl-file policy.acl --auth no --max-queues-per-user 2      --log-to-file localq.log > qpiddq.port
+    LOCAL_PORTQ=`cat qpiddq.port`
 }
 
 stop_brokers() {
         $QPIDD_EXEC --no-module-dir -q --port $LOCAL_PORT
         $QPIDD_EXEC --no-module-dir -q --port $LOCAL_PORTI
         $QPIDD_EXEC --no-module-dir -q --port $LOCAL_PORTU
+        $QPIDD_EXEC --no-module-dir -q --port $LOCAL_PORTQ
 }
 
 test_loading_acl_from_absolute_path(){
@@ -59,20 +63,24 @@ if test -d ${PYTHON_DIR} ;  then
     rm -rf $DATA_DIR
     rm -rf $DATA_DIRI
     rm -rf $DATA_DIRU
+    rm -rf $DATA_DIRQ
     mkdir -p $DATA_DIR
     mkdir -p $DATA_DIRI
     mkdir -p $DATA_DIRU
+    mkdir -p $DATA_DIRQ
     cp $srcdir/policy.acl $DATA_DIR
     cp $srcdir/policy.acl $DATA_DIRI
     cp $srcdir/policy.acl $DATA_DIRU
+    cp $srcdir/policy.acl $DATA_DIRQ
     start_brokers
-    echo "Running acl tests using brokers on ports $LOCAL_PORT, $LOCAL_PORTI, and $LOCAL_PORTU"
-    $QPID_PYTHON_TEST -b localhost:$LOCAL_PORT -m acl -Dport-i=$LOCAL_PORTI -Dport-u=$LOCAL_PORTU || EXITCODE=1
+    echo "Running acl tests using brokers on ports $LOCAL_PORT, $LOCAL_PORTI, $LOCAL_PORTU, and $LOCAL_PORTQ"
+    $QPID_PYTHON_TEST -b localhost:$LOCAL_PORT -m acl -Dport-i=$LOCAL_PORTI -Dport-u=$LOCAL_PORTU -Dport-q=$LOCAL_PORTQ || EXITCODE=1
     stop_brokers || EXITCODE=1
     test_loading_acl_from_absolute_path || EXITCODE=1
     rm -rf $DATA_DIR
     rm -rf $DATA_DIRI
     rm -rf $DATA_DIRU
+    rm -rf $DATA_DIRQ
     exit $EXITCODE
 fi