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@1444302 13f79535-47bb-0310-9956-ffa450edef68

10 files changed, 566 insertions, 96 deletions

diff --git a/qpid/cpp/src/qpid/acl/Acl.cpp b/qpid/cpp/src/qpid/acl/Acl.cpp
index 61e0b56104..31ad9a38ac 100644
--- a/qpid/cpp/src/qpid/acl/Acl.cpp
+++ b/qpid/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);
diff --git a/qpid/cpp/src/qpid/acl/AclConnectionCounter.cpp b/qpid/cpp/src/qpid/acl/AclConnectionCounter.cpp
index 195d8bee28..875137bf55 100644
--- a/qpid/cpp/src/qpid/acl/AclConnectionCounter.cpp
+++ b/qpid/cpp/src/qpid/acl/AclConnectionCounter.cpp
@@ -85,32 +85,32 @@ bool ConnectionCounter::limitApproveLH(
 //
 // countConnectionLH
 //
-// Increment the name's count in map and return a comparison against the limit.
-// called with dataLock already taken
+// Increment the name's count in map and return an optional comparison
+//  against a connection limit.
+// Called with dataLock already taken.
 //
 bool ConnectionCounter::countConnectionLH(
     connectCountsMap_t& theMap,
     const std::string& theName,
     uint16_t theLimit,
-    bool emitLog) {
+    bool emitLog,
+    bool enforceLimit) {
 
     bool result(true);
     uint16_t count(0);
-    if (theLimit > 0) {
-        connectCountsMap_t::iterator eRef = theMap.find(theName);
-        if (eRef != theMap.end()) {
-            count = (uint16_t)(*eRef).second + 1;
-            (*eRef).second = count;
-            result = count <= theLimit;
-        } else {
-            theMap[theName] = count = 1;
-        }
-        if (emitLog) {
-            QPID_LOG(trace, "ACL ConnectionApprover user=" << theName
-                << " limit=" << theLimit
-                << " curValue=" << count
-                << " result=" << (result ? "allow" : "deny"));
-        }
+    connectCountsMap_t::iterator eRef = theMap.find(theName);
+    if (eRef != theMap.end()) {
+        count = (uint16_t)(*eRef).second + 1;
+        (*eRef).second = count;
+        result = (enforceLimit ? count <= theLimit : true);
+    } else {
+        theMap[theName] = count = 1;
+    }
+    if (emitLog) {
+        QPID_LOG(trace, "ACL ConnectionApprover user=" << theName
+            << " limit=" << theLimit
+            << " curValue=" << count
+            << " result=" << (result ? "allow" : "deny"));
     }
     return result;
 }
@@ -123,23 +123,21 @@ bool ConnectionCounter::countConnectionLH(
 // called with dataLock already taken
 //
 void ConnectionCounter::releaseLH(
-    connectCountsMap_t& theMap, const std::string& theName, uint16_t theLimit) {
-
-    if (theLimit > 0) {
-        connectCountsMap_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;
-            }
+    connectCountsMap_t& theMap, const std::string& theName) {
+
+    connectCountsMap_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 {
-            // User had no connections.
-            QPID_LOG(notice, "ACL ConnectionCounter Connection for '" << theName
-                << "' not found in connection count pool");
+            (*eRef).second = count - 1;
         }
+    } else {
+        // User had no connections.
+        QPID_LOG(notice, "ACL ConnectionCounter Connection for '" << theName
+            << "' not found in connection count pool");
     }
 }
 
@@ -161,7 +159,7 @@ void ConnectionCounter::connection(broker::Connection& connection) {
     connectProgressMap[connection.getMgmtId()] = C_CREATED;
 
     // Count the connection from this host.
-    (void) countConnectionLH(connectByHostMap, hostName, hostLimit, false);
+    (void) countConnectionLH(connectByHostMap, hostName, hostLimit, false, false);
 }
 
 
@@ -180,8 +178,7 @@ void ConnectionCounter::closed(broker::Connection& connection) {
             // Normal case: connection was created and opened.
             // Decrement user in-use counts
             releaseLH(connectByNameMap,
-                      connection.getUserId(),
-                      nameLimit);
+                      connection.getUserId());
         } else {
             // Connection was created but not opened.
             // Don't decrement user count.
@@ -189,8 +186,7 @@ void ConnectionCounter::closed(broker::Connection& connection) {
 
         // Decrement host in-use count.
         releaseLH(connectByHostMap,
-                  getClientHost(connection.getMgmtId()),
-                  hostLimit);
+                  getClientHost(connection.getMgmtId()));
 
         // destroy connection progress indicator
         connectProgressMap.erase(eRef);
@@ -211,7 +207,10 @@ void ConnectionCounter::closed(broker::Connection& connection) {
 //  check total connections, connections from IP, connections by user and
 //  disallow if over any limit
 //
-bool ConnectionCounter::approveConnection(const broker::Connection& connection)
+bool ConnectionCounter::approveConnection(
+        const broker::Connection& connection,
+        bool enforcingConnectionQuotas,
+        uint16_t connectionUserQuota )
 {
     const std::string& hostName(getClientHost(connection.getMgmtId()));
     const std::string& userName(              connection.getUserId());
@@ -220,7 +219,7 @@ bool ConnectionCounter::approveConnection(const broker::Connection& connection)
 
     // Bump state from CREATED to OPENED
     (void) countConnectionLH(connectProgressMap, connection.getMgmtId(),
-                             C_OPENED, false);
+                             C_OPENED, false, false);
 
     // Approve total connections
     bool okTotal  = true;
@@ -235,7 +234,9 @@ bool ConnectionCounter::approveConnection(const broker::Connection& connection)
     bool okByIP   = limitApproveLH(connectByHostMap, hostName, hostLimit, true);
 
     // Count and Approve the connection by the user
-    bool okByUser = countConnectionLH(connectByNameMap, userName, nameLimit, true);
+    bool okByUser = countConnectionLH(connectByNameMap, userName,
+                                      connectionUserQuota, true,
+                                      enforcingConnectionQuotas);
 
     // Emit separate log for each disapproval
     if (!okTotal) {
@@ -252,7 +253,7 @@ bool ConnectionCounter::approveConnection(const broker::Connection& connection)
     }
     if (!okByUser) {
         QPID_LOG(error, "Client max per-user connection count limit of "
-                 << nameLimit << " exceeded by '"
+                 << connectionUserQuota << " exceeded by '"
                  << connection.getMgmtId() << "', user: '"
                  << userName << "'. Connection refused.");
     }
diff --git a/qpid/cpp/src/qpid/acl/AclConnectionCounter.h b/qpid/cpp/src/qpid/acl/AclConnectionCounter.h
index eec8e90256..e8ef35c1ba 100644
--- a/qpid/cpp/src/qpid/acl/AclConnectionCounter.h
+++ b/qpid/cpp/src/qpid/acl/AclConnectionCounter.h
@@ -77,12 +77,12 @@ private:
     bool countConnectionLH(connectCountsMap_t& theMap,
                            const std::string& theName,
                            uint16_t theLimit,
-                           bool emitLog);
+                           bool emitLog,
+                           bool enforceLimit);
 
     /** Release a connection */
     void releaseLH(connectCountsMap_t& theMap,
-                   const std::string& theName,
-                   uint16_t theLimit);
+                   const std::string& theName);
 
 public:
     ConnectionCounter(Acl& acl, uint16_t nl, uint16_t hl, uint16_t tl);
@@ -93,7 +93,9 @@ public:
     void     closed(broker::Connection& connection);
 
     // Connection counting
-    bool approveConnection(const broker::Connection& conn);
+    bool approveConnection(const broker::Connection& conn,
+                           bool enforcingConnectionQuotas,
+                           uint16_t connectionLimit );
 };
 
 }} // namespace qpid::ha
diff --git a/qpid/cpp/src/qpid/acl/AclData.cpp b/qpid/cpp/src/qpid/acl/AclData.cpp
index 997fbdf2eb..ca866ab7d3 100644
--- a/qpid/cpp/src/qpid/acl/AclData.cpp
+++ b/qpid/cpp/src/qpid/acl/AclData.cpp
@@ -25,11 +25,19 @@ namespace qpid {
 namespace acl {
 
     //
-    // Instantiate the substitution keyword string
+    // Instantiate the keyword strings
     //
-    const std::string AclData::USER_SUBSTITUTION_KEYWORD       = "${user}";
-    const std::string AclData::DOMAIN_SUBSTITUTION_KEYWORD     = "${domain}";
-    const std::string AclData::USERDOMAIN_SUBSTITUTION_KEYWORD = "${userdomain}";
+    const std::string AclData::ACL_KEYWORD_USER_SUBST        = "${user}";
+    const std::string AclData::ACL_KEYWORD_DOMAIN_SUBST      = "${domain}";
+    const std::string AclData::ACL_KEYWORD_USERDOMAIN_SUBST  = "${userdomain}";
+    const std::string AclData::ACL_KEYWORD_ALL               = "all";
+    const std::string AclData::ACL_KEYWORD_ACL               = "acl";
+    const std::string AclData::ACL_KEYWORD_GROUP             = "group";
+    const std::string AclData::ACL_KEYWORD_QUOTA             = "quota";
+    const std::string AclData::ACL_KEYWORD_QUOTA_CONNECTIONS = "connections";
+    const char        AclData::ACL_SYMBOL_WILDCARD           = '*';
+    const std::string AclData::ACL_KEYWORD_WILDCARD          = "*";
+    const char        AclData::ACL_SYMBOL_LINE_CONTINUATION  = '\\';
 
     //
     // constructor
@@ -37,7 +45,9 @@ namespace acl {
     AclData::AclData():
         decisionMode(qpid::acl::DENY),
         transferAcl(false),
-        aclSource("UNKNOWN")
+        aclSource("UNKNOWN"),
+        connQuotaRulesExist(false),
+        connQuotaRuleSettings(new quotaRuleSet)
     {
         for (unsigned int cnt=0; cnt< qpid::acl::ACTIONSIZE; cnt++)
         {
@@ -60,6 +70,9 @@ namespace acl {
             }
             delete[] actionList[cnt];
         }
+        transferAcl = false;
+        connQuotaRulesExist = false;
+        connQuotaRuleSettings->clear();
     }
 
 
@@ -73,7 +86,7 @@ namespace acl {
                             const std::string& lookupStr)
     {
         // allow wildcard on the end of rule strings...
-        if (ruleStr.data()[ruleStr.size()-1]=='*')
+        if (ruleStr.data()[ruleStr.size()-1]==ACL_SYMBOL_WILDCARD)
         {
             return ruleStr.compare(0,
                                    ruleStr.size()-1,
@@ -124,7 +137,7 @@ namespace acl {
 
             // If individual actorId not found then find a rule set for '*'.
             if (itrRule == actionList[action][objType]->end())
-                itrRule = actionList[action][objType]->find("*");
+                itrRule = actionList[action][objType]->find(ACL_KEYWORD_WILDCARD);
 
             if (itrRule != actionList[action][objType]->end())
             {
@@ -390,7 +403,7 @@ namespace acl {
             AclData::actObjItr itrRule = actionList[action][objType]->find(id);
 
             if (itrRule == actionList[action][objType]->end())
-                itrRule = actionList[action][objType]->find("*");
+                itrRule = actionList[action][objType]->find(ACL_KEYWORD_WILDCARD);
 
             if (itrRule != actionList[action][objType]->end() )
             {
@@ -436,9 +449,9 @@ namespace acl {
 
                     if (match && rsItr->pubRoutingKeyInRule)
                     {
-                        if ((routingKey.find(USER_SUBSTITUTION_KEYWORD, 0)       != std::string::npos) ||
-                            (routingKey.find(DOMAIN_SUBSTITUTION_KEYWORD, 0)     != std::string::npos) ||
-                            (routingKey.find(USERDOMAIN_SUBSTITUTION_KEYWORD, 0) != std::string::npos))
+                        if ((routingKey.find(ACL_KEYWORD_USER_SUBST, 0)       != std::string::npos) ||
+                            (routingKey.find(ACL_KEYWORD_DOMAIN_SUBST, 0)     != std::string::npos) ||
+                            (routingKey.find(ACL_KEYWORD_USERDOMAIN_SUBST, 0) != std::string::npos))
                         {
                             // The user is not allowed to present a routing key with the substitution key in it
                             QPID_LOG(debug, "ACL: Rule: " << rsItr->rawRuleNum <<
@@ -489,6 +502,62 @@ namespace acl {
     }
 
 
+
+    //
+    //
+    //
+    void AclData::setConnQuotaRuleSettings (
+                bool rulesExist, boost::shared_ptr<quotaRuleSet> quotaPtr)
+    {
+        connQuotaRulesExist = rulesExist;
+        connQuotaRuleSettings = quotaPtr;
+    }
+
+
+    //
+    // getConnQuotaForUser
+    //
+    // Return the true or false value of connQuotaRulesExist,
+    //  indicating whether any kind of lookup was done or not.
+    //
+    // When lookups are performed return the result value of
+    //  1. The user's setting else
+    //  2. The 'all' user setting else
+    //  3. Zero
+    // When lookups are not performed then return a result value of Zero.
+    //
+    bool AclData::getConnQuotaForUser(const std::string& theUserName,
+                                      uint16_t* theResult) const {
+        if (connQuotaRulesExist) {
+            // look for this user explicitly
+            quotaRuleSetItr nameItr = (*connQuotaRuleSettings).find(theUserName);
+            if (nameItr != (*connQuotaRuleSettings).end()) {
+                QPID_LOG(trace, "ACL: Connection quota for user " << theUserName
+                    << " explicitly set to : " << (*nameItr).second);
+                *theResult = (*nameItr).second;
+            } else {
+                // Look for the 'all' user
+                nameItr = (*connQuotaRuleSettings).find(ACL_KEYWORD_ALL);
+                if (nameItr != (*connQuotaRuleSettings).end()) {
+                    QPID_LOG(trace, "ACL: Connection quota for user " << theUserName
+                        << " chosen through value for 'all' : " << (*nameItr).second);
+                    *theResult = (*nameItr).second;
+                } else {
+                    // Neither userName nor "all" found.
+                    QPID_LOG(trace, "ACL: Connection quota for user " << theUserName
+                        << " absent in quota settings. Return value : 0");
+                    *theResult = 0;
+                }
+            }
+        } else {
+            // Rules do not exist
+            QPID_LOG(trace, "ACL: Connection quota for user " << theUserName
+                << " unavailable; quota settings are not specified. Return value : 0");
+            *theResult = 0;
+        }
+        return connQuotaRulesExist;
+    }
+
     //
     //
     //
@@ -656,9 +725,9 @@ namespace acl {
             domain = normalizeUserId(userId.substr(locDomSeparator+1));
         }
 
-        substituteString(ruleString, USER_SUBSTITUTION_KEYWORD,       user);
-        substituteString(ruleString, DOMAIN_SUBSTITUTION_KEYWORD,     domain);
-        substituteString(ruleString, USERDOMAIN_SUBSTITUTION_KEYWORD, userdomain);
+        substituteString(ruleString, ACL_KEYWORD_USER_SUBST,       user);
+        substituteString(ruleString, ACL_KEYWORD_DOMAIN_SUBST,     domain);
+        substituteString(ruleString, ACL_KEYWORD_USERDOMAIN_SUBST, userdomain);
     }
 
 
@@ -689,8 +758,8 @@ namespace acl {
             domain = normalizeUserId(userId.substr(locDomSeparator+1));
         }
         std::string oRule(ruleString);
-        substituteString(ruleString, userdomain, USERDOMAIN_SUBSTITUTION_KEYWORD);
-        substituteString(ruleString, user,       USER_SUBSTITUTION_KEYWORD);
-        substituteString(ruleString, domain,     DOMAIN_SUBSTITUTION_KEYWORD);
+        substituteString(ruleString, userdomain, ACL_KEYWORD_USERDOMAIN_SUBST);
+        substituteString(ruleString, user,       ACL_KEYWORD_USER_SUBST);
+        substituteString(ruleString, domain,     ACL_KEYWORD_DOMAIN_SUBST);
     }
 }}
diff --git a/qpid/cpp/src/qpid/acl/AclData.h b/qpid/cpp/src/qpid/acl/AclData.h
index b4b13c44b6..43cb5193f5 100644
--- a/qpid/cpp/src/qpid/acl/AclData.h
+++ b/qpid/cpp/src/qpid/acl/AclData.h
@@ -111,6 +111,8 @@ public:
     typedef  std::map<std::string, ruleSet > actionObject; // user
     typedef  actionObject::iterator          actObjItr;
     typedef  actionObject*                   aclAction;
+    typedef  std::map<std::string, uint16_t> quotaRuleSet; // <username, N>
+    typedef  quotaRuleSet::const_iterator    quotaRuleSetItr;
 
     // Action*[] -> Object*[] -> map<user -> set<Rule> >
     aclAction*           actionList[qpid::acl::ACTIONSIZE];
@@ -134,9 +136,18 @@ public:
 
     bool matchProp(const std::string & src, const std::string& src1);
     void clear ();
-    static const std::string USER_SUBSTITUTION_KEYWORD;
-    static const std::string DOMAIN_SUBSTITUTION_KEYWORD;
-    static const std::string USERDOMAIN_SUBSTITUTION_KEYWORD;
+    static const std::string ACL_KEYWORD_USER_SUBST;
+    static const std::string ACL_KEYWORD_DOMAIN_SUBST;
+    static const std::string ACL_KEYWORD_USERDOMAIN_SUBST;
+    static const std::string ACL_KEYWORD_ALL;
+    static const std::string ACL_KEYWORD_ACL;
+    static const std::string ACL_KEYWORD_GROUP;
+    static const std::string ACL_KEYWORD_QUOTA;
+    static const std::string ACL_KEYWORD_QUOTA_CONNECTIONS;
+    static const char        ACL_SYMBOL_WILDCARD;
+    static const std::string ACL_KEYWORD_WILDCARD;
+    static const char        ACL_SYMBOL_LINE_CONTINUATION;
+
     void substituteString(std::string& targetString,
                           const std::string& placeholder,
                           const std::string& replacement);
@@ -146,6 +157,31 @@ public:
     void substituteKeywords(std::string& ruleString,
                             const std::string& userId);
 
+    // Per user connection quotas extracted from acl rule file
+    //   Set by reader
+    void setConnQuotaRuleSettings (bool, boost::shared_ptr<quotaRuleSet>);
+    //   Get by connection approvers
+    bool enforcingConnectionQuotas() { return connQuotaRulesExist; }
+    bool getConnQuotaForUser(const std::string&, uint16_t*) const;
+
+    /** getConnectMaxSpec
+     * Connection quotas are held in uint16_t variables.
+     * This function specifies the largest value that a user is allowed
+     * to declare for a connection quota. The upper limit serves two
+     * purposes: 1. It leaves room for magic numbers that may be declared
+     * by keyword names in Acl files and not have those numbers conflict
+     * with innocent user declared values, and 2. It makes the unsigned
+     * math very close to _MAX work reliably with no risk of accidental
+     * wrapping back to zero.
+     */
+    static uint16_t getConnectMaxSpec() {
+        return 65530;
+    }
+    static std::string getMaxConnectSpecStr() {
+        return "65530";
+    }
+
+
     AclData();
     virtual ~AclData();
 
@@ -157,6 +193,10 @@ private:
     bool compareIntMin(const qpid::acl::SpecProperty theProperty,
                        const std::string             theAclValue,
                        const std::string             theLookupValue);
+
+    // Per-user connection quota
+    bool connQuotaRulesExist;
+    boost::shared_ptr<quotaRuleSet> connQuotaRuleSettings; // Map of user-to-N values from rule file
 };
 
 }} // namespace qpid::acl
diff --git a/qpid/cpp/src/qpid/acl/AclPlugin.cpp b/qpid/cpp/src/qpid/acl/AclPlugin.cpp
index 64bf040309..c666eb5420 100644
--- a/qpid/cpp/src/qpid/acl/AclPlugin.cpp
+++ b/qpid/cpp/src/qpid/acl/AclPlugin.cpp
@@ -42,8 +42,8 @@ struct AclOptions : public Options {
         values.aclMaxConnectTotal = 500;
         addOptions()
             ("acl-file",           optValue(values.aclFile, "FILE"), "The policy file to load from, loaded from data dir")
-            ("max-connections"         , optValue(values.aclMaxConnectTotal, "N"),   "The maximum combined number of connections allowed. 0 implies no limit.")
             ("connection-limit-per-user", optValue(values.aclMaxConnectPerUser, "N"), "The maximum number of connections allowed per user. 0 implies no limit.")
+            ("max-connections"         , optValue(values.aclMaxConnectTotal, "N"),   "The maximum combined number of connections allowed. 0 implies no limit.")
             ("connection-limit-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/qpid/cpp/src/qpid/acl/AclReader.cpp b/qpid/cpp/src/qpid/acl/AclReader.cpp
index 3d5a6662db..7eb9b82c64 100644
--- a/qpid/cpp/src/qpid/acl/AclReader.cpp
+++ b/qpid/cpp/src/qpid/acl/AclReader.cpp
@@ -24,6 +24,7 @@
 #include <sstream>
 #include "qpid/log/Statement.h"
 #include "qpid/Exception.h"
+#include <boost/lexical_cast.hpp>
 
 #include <iomanip> // degug
 #include <iostream> // debug
@@ -95,7 +96,7 @@ namespace acl {
                 << cnt << " " << (*i)->toString());
 
             if (!foundmode && (*i)->actionAll && (*i)->names.size() == 1
-                && (*((*i)->names.begin())).compare("*") == 0) {
+                && (*((*i)->names.begin())).compare(AclData::ACL_KEYWORD_WILDCARD) == 0) {
                     d->decisionMode = (*i)->res;
                     QPID_LOG(debug, "ACL: FoundMode "
                         << AclHelper::getAclResultStr(d->decisionMode));
@@ -105,9 +106,9 @@ namespace acl {
 
                 // Record which properties have the user substitution string
                 for (pmCitr pItr=rule.props.begin(); pItr!=rule.props.end(); pItr++) {
-                    if ((pItr->second.find(AclData::USER_SUBSTITUTION_KEYWORD, 0)       != std::string::npos) ||
-                        (pItr->second.find(AclData::DOMAIN_SUBSTITUTION_KEYWORD, 0)     != std::string::npos) ||
-                        (pItr->second.find(AclData::USERDOMAIN_SUBSTITUTION_KEYWORD, 0) != std::string::npos)) {
+                    if ((pItr->second.find(AclData::ACL_KEYWORD_USER_SUBST, 0)       != std::string::npos) ||
+                        (pItr->second.find(AclData::ACL_KEYWORD_DOMAIN_SUBST, 0)     != std::string::npos) ||
+                        (pItr->second.find(AclData::ACL_KEYWORD_USERDOMAIN_SUBST, 0) != std::string::npos)) {
                         rule.ruleHasUserSub[pItr->first] = true;
                     }
                 }
@@ -167,7 +168,7 @@ namespace acl {
                         // add users and Rule to object set
                         bool allNames = false;
                         // check to see if names.begin is '*'
-                        if ((*(*i)->names.begin()).compare("*") == 0)
+                        if ((*(*i)->names.begin()).compare(AclData::ACL_KEYWORD_WILDCARD) == 0)
                             allNames = true;
 
                         for (nsCitr itr  = (allNames ? names.begin() : (*i)->names.begin());
@@ -199,7 +200,7 @@ namespace acl {
                         objstr << AclHelper::getObjectTypeStr((ObjectType) ocnt) << ",";
                 }
 
-                bool allNames = ((*(*i)->names.begin()).compare("*") == 0);
+                bool allNames = ((*(*i)->names.begin()).compare(AclData::ACL_KEYWORD_WILDCARD) == 0);
                 std::ostringstream userstr;
                 for (nsCitr itr  = (allNames ? names.begin() : (*i)->names.begin());
                             itr != (allNames ? names.end()   : (*i)->names.end());
@@ -218,12 +219,15 @@ namespace acl {
                     << "}" );
             }
         }
+
+        // connection quota
+        d->setConnQuotaRuleSettings(connQuotaRulesExist, connQuota);
     }
 
 
     void AclReader::aclRule::processName(const std::string& name, const groupMap& groups) {
-        if (name.compare("all") == 0) {
-            names.insert("*");
+        if (name.compare(AclData::ACL_KEYWORD_ALL) == 0) {
+            names.insert(AclData::ACL_KEYWORD_WILDCARD);
         } else {
             gmCitr itr = groups.find(name);
             if (itr == groups.end()) {
@@ -234,9 +238,13 @@ namespace acl {
         }
     }
 
-    AclReader::AclReader() : lineNumber(0), contFlag(false), validationMap(new AclHelper::objectMap) {
+    AclReader::AclReader(uint16_t theCliMaxConnPerUser) : lineNumber(0), contFlag(false),
+        validationMap(new AclHelper::objectMap),
+        cliMaxConnPerUser (theCliMaxConnPerUser),
+        connQuotaRulesExist(false),
+        connQuota(new AclData::quotaRuleSet) {
         AclHelper::loadValidationMap(validationMap);
-        names.insert("*");
+        names.insert(AclData::ACL_KEYWORD_WILDCARD);
     }
 
     AclReader::~AclReader() {}
@@ -254,6 +262,11 @@ namespace acl {
             errorStream << "Unable to open ACL file \"" << fn << "\": eof=" << (ifs.eof()?"T":"F") << "; fail=" << (ifs.fail()?"T":"F") << "; bad=" << (ifs.bad()?"T":"F");
             return -1;
         }
+        // Propagate nonzero per-user max connection setting from CLI
+        if (cliMaxConnPerUser > 0) {
+            (*connQuota)[AclData::ACL_KEYWORD_ACL] = cliMaxConnPerUser;
+        }
+        // Loop to process the Acl file
         try {
             bool err = false;
             while (ifs.good()) {
@@ -282,6 +295,7 @@ namespace acl {
         }
         printNames();
         printRules();
+        printConnectionQuotas();
         loadDecisionData(d);
 
         return 0;
@@ -292,7 +306,7 @@ namespace acl {
         std::vector<std::string> toks;
 
         // Check for continuation
-        char* contCharPtr = std::strrchr(line, '\\');
+        char* contCharPtr = std::strrchr(line, AclData::ACL_SYMBOL_LINE_CONTINUATION);
         bool cont = contCharPtr != 0;
         if (cont) *contCharPtr = 0;
 
@@ -303,10 +317,12 @@ namespace acl {
             return false;
         }
 
-        if (numToks && (toks[0].compare("group") == 0 || contFlag)) {
+        if (numToks && (toks[0].compare(AclData::ACL_KEYWORD_GROUP) == 0 || contFlag)) {
             ret = processGroupLine(toks, cont);
-        } else if (numToks && toks[0].compare("acl") == 0) {
+        } else if (numToks && toks[0].compare(AclData::ACL_KEYWORD_ACL) == 0) {
             ret = processAclLine(toks);
+        } else if (numToks && toks[0].compare(AclData::ACL_KEYWORD_QUOTA) == 0) {
+            ret = processQuotaLine(toks);
         } else {
             // Check for whitespace only line, ignore these
             bool ws = true;
@@ -317,7 +333,10 @@ namespace acl {
                 ret = true;
             } else {
                 errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber
-                    << ", Non-continuation line must start with \"group\" or \"acl\".";
+                    << ", Non-continuation line must start with \""
+                    << AclData::ACL_KEYWORD_GROUP << "\", \""
+                    << AclData::ACL_KEYWORD_ACL << "\". or \""
+                    << AclData::ACL_KEYWORD_QUOTA << "\".";
                 ret = false;
             }
         }
@@ -337,6 +356,102 @@ namespace acl {
         return cnt;
     }
 
+
+    // Process 'quota' rule lines
+    // Return true if the line is successfully processed without errors
+    bool AclReader::processQuotaLine(tokList& toks) {
+        const unsigned toksSize = toks.size();
+        const unsigned minimumSize = 3;
+        if (toksSize < minimumSize) {
+            errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber
+                << ", Insufficient tokens for quota definition.";
+            return false;
+        }
+
+        if (toks[1].compare(AclData::ACL_KEYWORD_QUOTA_CONNECTIONS) == 0) {
+            return processQuotaConnLine(toks);
+        } else {
+            errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber
+                << ", Quota type \"" << toks[1] << "\" unrecognized.";
+            return false;
+        }
+    }
+
+
+    // Process 'quota connections' rule lines
+    // Return true if the line is successfully processed without errors
+    bool AclReader::processQuotaConnLine(tokList& toks) {
+        const unsigned toksSize = toks.size();
+
+        uint16_t nConns(0);
+        try {
+            nConns = boost::lexical_cast<uint16_t>(toks[2]);
+        } catch(const boost::bad_lexical_cast&) {
+            errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber
+                << ", Connection quota value \"" << toks[2]
+                << "\" cannot be converted to a 16-bit unsigned integer.";
+            return false;
+        }
+
+        // limit check the connection setting
+        if (nConns > AclData::getConnectMaxSpec())
+        {
+            errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber
+                << ", Connection quota value \"" << toks[2]
+                << "\" exceeds maximum configuration setting of "
+                << AclData::getConnectMaxSpec();
+            return false;
+        }
+
+        // Apply the connection count to all names in rule
+        for (unsigned idx = 3; idx < toksSize; idx++) {
+            if (groups.find(toks[idx]) == groups.end()) {
+                // This is the name of an individual, not a group
+                (*connQuota)[toks[idx]] = nConns;
+            } else {
+                if (!processQuotaConnGroup(toks[idx], nConns))
+                    return false;
+            }
+        }
+        return true;
+    }
+
+
+    // Process 'quota connections' group expansion
+    // Return true if the quota is applied to all members of the group
+    bool AclReader::processQuotaConnGroup(const std::string& theGroup, uint16_t theQuota) {
+        gmCitr citr = groups.find(theGroup);
+
+        if (citr == groups.end()) {
+            errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber
+                << ", Failed to expand group \"" << theGroup << "\".";
+            return false;
+        }
+
+        for (nsCitr gni=citr->second->begin(); gni!=citr->second->end(); gni++) {
+            if (groups.find(*gni) == groups.end()) {
+                (*connQuota)[*gni] = theQuota;
+            } else {
+                if (!processQuotaConnGroup(*gni, theQuota))
+                    return false;
+            }
+        }
+        return true;
+    }
+
+
+    void AclReader::printConnectionQuotas() const {
+        QPID_LOG(debug, "ACL: connection quota: " << (*connQuota).size() << " rules found:");
+        int cnt = 1;
+        for (AclData::quotaRuleSetItr itr=(*connQuota).begin();
+                                      itr != (*connQuota).end();
+                                      ++itr,++cnt) {
+            QPID_LOG(debug, "ACL: quota " << cnt << " : " << (*itr).second
+                << " connections for " << (*itr).first)
+        }
+    }
+
+
     // Return true if the line is successfully processed without errors
     // If cont is true, then groupName must be set to the continuation group name
     bool AclReader::processGroupLine(tokList& toks, const bool cont) {
@@ -462,8 +577,8 @@ namespace acl {
             return false;
         }
 
-        bool actionAllFlag = toks[3].compare("all") == 0;
-        bool userAllFlag   = toks[2].compare("all") == 0;
+        bool actionAllFlag = toks[3].compare(AclData::ACL_KEYWORD_ALL) == 0;
+        bool userAllFlag   = toks[2].compare(AclData::ACL_KEYWORD_ALL) == 0;
         Action action;
         if (actionAllFlag) {
 
@@ -492,7 +607,7 @@ namespace acl {
         }
 
         if (toksSize >= 5) { // object name-value pair
-            if (toks[4].compare("all") == 0) {
+            if (toks[4].compare(AclData::ACL_KEYWORD_ALL) == 0) {
                 rule->setObjectTypeAll();
             } else {
                 try {
@@ -526,7 +641,7 @@ namespace acl {
             }
         }
         // Check if name (toks[2]) is group; if not, add as name of individual
-        if (toks[2].compare("all") != 0) {
+        if (toks[2].compare(AclData::ACL_KEYWORD_ALL) != 0) {
             if (groups.find(toks[2]) == groups.end()) {
                 addName(toks[2]);
             }
diff --git a/qpid/cpp/src/qpid/acl/AclReader.h b/qpid/cpp/src/qpid/acl/AclReader.h
index 6351c1e509..1fa374c59c 100644
--- a/qpid/cpp/src/qpid/acl/AclReader.h
+++ b/qpid/cpp/src/qpid/acl/AclReader.h
@@ -28,6 +28,7 @@
 #include <sstream>
 #include <memory>
 #include "qpid/acl/AclData.h"
+#include "qpid/acl/Acl.h"
 #include "qpid/broker/AclModule.h"
 
 namespace qpid {
@@ -96,7 +97,7 @@ class AclReader {
     std::ostringstream      errorStream;
 
   public:
-    AclReader();
+    AclReader(uint16_t cliMaxConnPerUser);
     virtual ~AclReader();
     int read(const std::string& fn, boost::shared_ptr<AclData> d); // return=0 for success
     std::string getError();
@@ -116,8 +117,17 @@ class AclReader {
     void printRules() const; // debug aid
     bool isValidUserName(const std::string& name);
 
+    bool processQuotaLine(tokList& toks);
+    bool processQuotaConnLine(tokList& toks);
+    bool processQuotaConnGroup(const std::string&, uint16_t);
+    void printConnectionQuotas() const;
+
     static bool isValidGroupName(const std::string& name);
     static nvPair splitNameValuePair(const std::string& nvpString);
+
+    const uint16_t cliMaxConnPerUser;
+    bool connQuotaRulesExist;
+    boost::shared_ptr<AclData::quotaRuleSet> connQuota;
 };
 
 }} // namespace qpid::acl
diff --git a/qpid/cpp/src/qpid/acl/AclTopicMatch.h b/qpid/cpp/src/qpid/acl/AclTopicMatch.h
index 486c229ad5..654d1d63d4 100644
--- a/qpid/cpp/src/qpid/acl/AclTopicMatch.h
+++ b/qpid/cpp/src/qpid/acl/AclTopicMatch.h
@@ -30,7 +30,7 @@ namespace qpid {
 namespace broker {
 
 // Class for executing topic exchange routing key matching rules in
-// Acl code the allows or denies users publishing to an exchange.
+// Acl code. Allows or denies users publishing to an exchange.
 class TopicExchange::TopicExchangeTester {
 
 class boundNode;
diff --git a/qpid/cpp/src/tests/acl.py b/qpid/cpp/src/tests/acl.py
index 1020a2eff6..48723bfde9 100755
--- a/qpid/cpp/src/tests/acl.py
+++ b/qpid/cpp/src/tests/acl.py
@@ -2065,36 +2065,242 @@ class ACLTests(TestBase010):
    # Connection limits
    #=====================================
 
-    def test_connection_limits(self):
+    def test_connection_limits_cli_sets_all(self):
+
+        try:
+            sessiona1 = self.get_session_by_port('alice','alice', self.port_u())
+            sessiona2 = self.get_session_by_port('alice','alice', self.port_u())
+        except Exception, e:
+            self.fail("Could not create two connections for user alice: " + str(e))
+
+        # Third session should fail
+        try:
+            sessiona3 = self.get_session_by_port('alice','alice', self.port_u())
+            self.fail("Should not be able to create third connection for user alice")
+        except Exception, e:
+            result = None
+
+
+
+    def test_connection_limits_by_named_user(self):
         """
         Test ACL control connection limits
         """
+        aclf = self.get_acl_file()
+        aclf.write('quota connections 2 alice bob\n')
+        aclf.write('quota connections 0 evildude\n')
+        aclf.write('acl allow all all')
+        aclf.close()
+
+        result = self.reload_acl()
+        if (result):
+            self.fail(result)
+
         # By username should be able to connect twice per user
         try:
-            sessiona1 = self.get_session_by_port('alice','alice', self.port_u())
-            sessiona2 = self.get_session_by_port('alice','alice', self.port_u())
+            sessiona1 = self.get_session('alice','alice')
+            sessiona2 = self.get_session('alice','alice')
         except Exception, e:
             self.fail("Could not create two connections for user alice: " + str(e))
 
         # Third session should fail
         try:
-            sessiona3 = self.get_session_by_port('alice','alice', self.port_u())
+            sessiona3 = self.get_session('alice','alice')
+            self.fail("Should not be able to create third connection for user alice")
+        except Exception, e:
+            result = None
+
+        # Disconnecting should allow another session.
+        sessiona1.close()
+        try:
+            sessiona3 = self.get_session('alice','alice')
+        except Exception, e:
+            self.fail("Could not recreate second connection for user alice: " + str(e))
+
+        # By username should be able to connect twice per user
+        try:
+            sessionb1 = self.get_session('bob','bob')
+            sessionb2 = self.get_session('bob','bob')
+        except Exception, e:
+            self.fail("Could not create two connections for user bob: " + str(e))
+
+        # Third session should fail
+        try:
+            sessionb3 = self.get_session('bob','bob')
+            self.fail("Should not be able to create third connection for user bob")
+        except Exception, e:
+            result = None
+
+
+        # User with quota of 0 is denied
+        try:
+            sessione1 = self.get_session('evildude','evildude')
+            self.fail("Should not be able to create a connection for user evildude")
+        except Exception, e:
+            result = None
+
+
+        # User not named in quotas is denied
+        try:
+            sessionc1 = self.get_session('charlie','charlie')
+            self.fail("Should not be able to create a connection for user charlie")
+        except Exception, e:
+            result = None
+
+        # Clean up the sessions
+        sessiona2.close()
+        sessiona3.close()
+        sessionb1.close()
+        sessionb2.close()
+
+
+
+    def test_connection_limits_by_unnamed_all(self):
+        """
+        Test ACL control connection limits
+        """
+        aclf = self.get_acl_file()
+        aclf.write('quota connections 2 alice bob\n')
+        aclf.write('quota connections 1 all\n')
+        aclf.write('acl allow all all')
+        aclf.close()
+
+        result = self.reload_acl()
+        if (result):
+            self.fail(result)
+
+        # By username should be able to connect twice per user
+        try:
+            sessiona1 = self.get_session('alice','alice')
+            sessiona2 = self.get_session('alice','alice')
+        except Exception, e:
+            self.fail("Could not create two connections for user alice: " + str(e))
+
+        # Third session should fail
+        try:
+            sessiona3 = self.get_session('alice','alice')
             self.fail("Should not be able to create third connection for user alice")
         except Exception, e:
             result = None
 
+        # By username should be able to connect twice per user
         try:
-            sessionb1 = self.get_session_by_port('bob','bob', self.port_u())
-            sessionb2 = self.get_session_by_port('bob','bob', self.port_u())
+            sessionb1 = self.get_session('bob','bob')
+            sessionb2 = self.get_session('bob','bob')
         except Exception, e:
             self.fail("Could not create two connections for user bob: " + str(e))
 
+        # Third session should fail
         try:
-            sessionb3 = self.get_session_by_port('bob','bob', self.port_u())
+            sessionb3 = self.get_session('bob','bob')
             self.fail("Should not be able to create third connection for user bob")
         except Exception, e:
             result = None
 
+        # User not named in quotas gets 'all' quota
+        try:
+            sessionc1 = self.get_session('charlie','charlie')
+        except Exception, e:
+            self.fail("Could not create one connection for user charlie: " + str(e))
+
+        # Next session should fail
+        try:
+            sessionc2 = self.get_session('charlie','charlie')
+            self.fail("Should not be able to create second connection for user charlie")
+        except Exception, e:
+            result = None
+
+        # Clean up the sessions
+        sessiona1.close()
+        sessiona2.close()
+        sessionb1.close()
+        sessionb2.close()
+        sessionc1.close()
+
+
+    def test_connection_limits_by_group(self):
+        """
+        Test ACL control connection limits
+        """
+        aclf = self.get_acl_file()
+        aclf.write('group stooges moe@QPID larry@QPID curly@QPID\n')
+        aclf.write('quota connections 2 alice bob\n')
+        aclf.write('quota connections 2 stooges charlie\n')
+        aclf.write('# user and groups may be overwritten. Should use last value\n')
+        aclf.write('quota connections 3 bob stooges\n')
+        aclf.write('acl allow all all')
+        aclf.close()
+
+        result = self.reload_acl()
+        if (result):
+            self.fail(result)
+
+        # Alice gets 2
+        try:
+            sessiona1 = self.get_session('alice','alice')
+            sessiona2 = self.get_session('alice','alice')
+        except Exception, e:
+            self.fail("Could not create two connections for user alice: " + str(e))
+
+        # Third session should fail
+        try:
+            sessiona3 = self.get_session('alice','alice')
+            self.fail("Should not be able to create third connection for user alice")
+        except Exception, e:
+            result = None
+
+        # Bob gets 3
+        try:
+            sessionb1 = self.get_session('bob','bob')
+            sessionb2 = self.get_session('bob','bob')
+            sessionb3 = self.get_session('bob','bob')
+        except Exception, e:
+            self.fail("Could not create three connections for user bob: " + str(e))
+
+        # Fourth session should fail
+        try:
+            sessionb4 = self.get_session('bob','bob')
+            self.fail("Should not be able to create fourth connection for user bob")
+        except Exception, e:
+            result = None
+
+        # Moe gets 3
+        try:
+            sessionm1 = self.get_session('moe','moe')
+            sessionm2 = self.get_session('moe','moe')
+            sessionm3 = self.get_session('moe','moe')
+        except Exception, e:
+            self.fail("Could not create three connections for user moe: " + str(e))
+
+        # Fourth session should fail
+        try:
+            sessionb4 = self.get_session('moe','moe')
+            self.fail("Should not be able to create fourth connection for user ,pe")
+        except Exception, e:
+            result = None
+
+        # User not named in quotas is denied
+        try:
+            sessions1 = self.get_session('shemp','shemp')
+            self.fail("Should not be able to create a connection for user shemp")
+        except Exception, e:
+            result = None
+
+        # Clean up the sessions
+        sessiona1.close()
+        sessiona2.close()
+        sessionb1.close()
+        sessionb2.close()
+        sessionb3.close()
+        sessionm1.close()
+        sessionm2.close()
+        sessionm3.close()
+
+
+    def test_connection_limits_by_ip_address(self):
+        """
+        Test ACL control connection limits by ip address
+        """
         # By IP address should be able to connect twice per client address
         try:
             sessionb1 = self.get_session_by_port('alice','alice', self.port_i())
@@ -2109,6 +2315,8 @@ class ACLTests(TestBase010):
         except Exception, e:
             result = None
 
+        sessionb1.close()
+        sessionb2.close()
 
    #=====================================
    # User name substitution