QPID-4054 C++ Broker connection limits per user

1. Constrain maximum limits to be a few ticks below Uint16_t max to avoid inadvertent wrapping and to allow room for some named constants such as UNLIMITED.
2. Add syntax to Acl rule file
   quota connections N user|group [user|group]
3. Pseudo user 'all' receives value from command line switch or from Acl rule file.
4. Named constant strings used in comparisons instead of local strings.
5. Connection counts maintained all the time to support reolad of Acl rule file that may change limits.
6. Self tests exercise all the features.



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

1 files changed, 27 insertions, 2 deletions

diff --git a/cpp/src/qpid/acl/Acl.cpp b/cpp/src/qpid/acl/Acl.cpp
index 61e0b56104..31ad9a38ac 100644
--- a/cpp/src/qpid/acl/Acl.cpp
+++ b/cpp/src/qpid/acl/Acl.cpp
@@ -24,6 +24,7 @@
 #include "qpid/sys/Mutex.h"
 
 #include "qpid/broker/Broker.h"
+#include "qpid/broker/Connection.h"
 #include "qpid/Plugin.h"
 #include "qpid/Options.h"
 #include "qpid/log/Logger.h"
@@ -56,6 +57,15 @@ Acl::Acl (AclValues& av, Broker& b): aclValues(av), broker(&b), transferAcl(fals
     connectionCounter(new ConnectionCounter(*this, aclValues.aclMaxConnectPerUser, aclValues.aclMaxConnectPerIp, aclValues.aclMaxConnectTotal)),
     resourceCounter(new ResourceCounter(*this, aclValues.aclMaxQueuesPerUser)){
 
+    if (aclValues.aclMaxConnectPerUser > AclData::getConnectMaxSpec())
+        throw Exception("--connection-limit-per-user switch cannot be larger than " + AclData::getMaxConnectSpecStr());
+    if (aclValues.aclMaxConnectPerIp > AclData::getConnectMaxSpec())
+        throw Exception("--connection-limit-per-ip switch cannot be larger than " + AclData::getMaxConnectSpecStr());
+    if (aclValues.aclMaxConnectTotal > AclData::getConnectMaxSpec())
+        throw Exception("--max-connections switch cannot be larger than " + AclData::getMaxConnectSpecStr());
+    if (aclValues.aclMaxQueuesPerUser > AclData::getConnectMaxSpec())
+        throw Exception("--max-queues-per-user switch cannot be larger than " + AclData::getMaxConnectSpecStr());
+
     agent = broker->getManagementAgent();
 
     if (agent != 0){
@@ -138,7 +148,18 @@ bool Acl::authorise(
 
 bool Acl::approveConnection(const qpid::broker::Connection& conn)
 {
-    return connectionCounter->approveConnection(conn);
+    const std::string& userName(conn.getUserId());
+    uint16_t connectionLimit(0);
+
+    boost::shared_ptr<AclData> dataLocal;
+    {
+        Mutex::ScopedLock locker(dataLock);
+        dataLocal = data;  //rcu copy
+    }
+
+    bool enforcingConnQuotas = dataLocal->getConnQuotaForUser(userName, &connectionLimit);
+
+    return connectionCounter->approveConnection(conn, enforcingConnQuotas, connectionLimit);
 }
 
 bool Acl::approveCreateQueue(const std::string& userId, const std::string& queueName)
@@ -207,7 +228,7 @@ bool Acl::readAclFile(std::string& errorText)
 
 bool Acl::readAclFile(std::string& aclFile, std::string& errorText) {
     boost::shared_ptr<AclData> d(new AclData);
-    AclReader ar;
+    AclReader ar(aclValues.aclMaxConnectPerUser);
     if (ar.read(aclFile, d)){
         agent->raiseEvent(_qmf::EventFileLoadFailed("", ar.getError()));
         errorText = ar.getError();
@@ -228,6 +249,10 @@ bool Acl::readAclFile(std::string& aclFile, std::string& errorText) {
         QPID_LOG(debug,"ACL: Transfer ACL is Enabled!");
     }
 
+    if (data->enforcingConnectionQuotas()){
+        QPID_LOG(debug, "ACL: Connection quotas are Enabled.");
+    }
+
     data->aclSource = aclFile;
     if (mgmtObject!=0){
         mgmtObject->set_transferAcl(transferAcl?1:0);