QPID-4123: C++ Broker ACL creates too many rules

Recent changes have added new tables to define what
are ACL lookups and their properties. This commit
finishes that work by not propagating rules that
will never match. Also, it completes the scaffolding
for allowed and denied host lists to be fully 
integrated. This commit:

* Adds startup logging of ACL validation tables
  with cross references to possible rule matches.
* Hooks the ACL host allow/deny connection lists
  into self test.
* Fixes self tests that get broken by proper rule
  table handling.
* Introduces a 'create connection' decision mode
  similar to ACL rule decision mode.
* Describes it all in doc book.



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

14 files changed, 759 insertions, 220 deletions

diff --git a/qpid/cpp/src/qpid/acl/Acl.cpp b/qpid/cpp/src/qpid/acl/Acl.cpp
index cc3a08c754..bd9482ef41 100644
--- a/qpid/cpp/src/qpid/acl/Acl.cpp
+++ b/qpid/cpp/src/qpid/acl/Acl.cpp
@@ -169,17 +169,13 @@ bool Acl::approveConnection(const qpid::broker::Connection& conn)
     }
 
     (void) dataLocal->getConnQuotaForUser(userName, &connectionLimit);
-    boost::shared_ptr<const AclData::bwHostRuleSet> globalRules = dataLocal->getGlobalConnectionRules();
-    boost::shared_ptr<const AclData::bwHostRuleSet> userRules   = dataLocal->getUserConnectionRules(userName);
 
-
-    return connectionCounter->approveConnection(conn,
-                                                userName,
-                                                dataLocal->enforcingConnectionQuotas(),
-                                                connectionLimit,
-                                                globalRules,
-                                                userRules
-                                               );
+    return connectionCounter->approveConnection(
+        conn,
+        userName,
+        dataLocal->enforcingConnectionQuotas(),
+        connectionLimit,
+        dataLocal);
 }
 
 bool Acl::approveCreateQueue(const std::string& userId, const std::string& queueName)
@@ -295,6 +291,9 @@ bool Acl::readAclFile(std::string& aclFile, std::string& errorText) {
         QPID_LOG(debug, "ACL: Queue quotas are Enabled.");
     }
 
+    QPID_LOG(debug, "ACL: Default connection mode : "
+        << AclHelper::getAclResultStr(d->connectionMode()));
+
     data->aclSource = aclFile;
     if (mgmtObject!=0){
         mgmtObject->set_transferAcl(transferAcl?1:0);
@@ -317,6 +316,7 @@ void Acl::loadEmptyAclRuleset() {
     boost::shared_ptr<AclData> d(new AclData);
     d->decisionMode = ALLOW;
     d->aclSource = "";
+    d->connectionDecisionMode = ALLOW;
     {
         Mutex::ScopedLock locker(dataLock);
         data = d;
@@ -357,13 +357,23 @@ Manageable::status_t Acl::lookup(qpid::management::Args& args, std::string& text
             Mutex::ScopedLock locker(dataLock);
             dataLocal = data;  //rcu copy
         }
-        AclResult aclResult = dataLocal->lookup(
-            ioArgs.i_userId,
-            action,
-            objType,
-            ioArgs.i_objectName,
-            &propertyMap);
-
+        AclResult aclResult;
+        // CREATE CONNECTION does not use lookup()
+        if (action == ACT_CREATE && objType == OBJ_CONNECTION) {
+            std::string host = propertyMap[acl::PROP_HOST];
+            std::string logString;
+            aclResult = dataLocal->isAllowedConnection(
+                ioArgs.i_userId,
+                host,
+                logString);
+        } else {
+            aclResult = dataLocal->lookup(
+                ioArgs.i_userId,
+                action,
+                objType,
+                ioArgs.i_objectName,
+                &propertyMap);
+        }
         ioArgs.o_result = AclHelper::getAclResultStr(aclResult);
         result = STATUS_OK;
 
diff --git a/qpid/cpp/src/qpid/acl/AclConnectionCounter.cpp b/qpid/cpp/src/qpid/acl/AclConnectionCounter.cpp
index 0e780a95bc..ca3da50088 100644
--- a/qpid/cpp/src/qpid/acl/AclConnectionCounter.cpp
+++ b/qpid/cpp/src/qpid/acl/AclConnectionCounter.cpp
@@ -139,8 +139,9 @@ void ConnectionCounter::releaseLH(
         }
     } else {
         // User had no connections.
-        QPID_LOG(notice, "ACL ConnectionCounter Connection for '" << theName
-            << "' not found in connection count pool");
+        // Connections denied by ACL never get users added
+        //QPID_LOG(notice, "ACL ConnectionCounter Connection for '" << theName
+        //    << "' not found in connection count pool");
     }
 }
 
@@ -215,8 +216,7 @@ bool ConnectionCounter::approveConnection(
         const std::string& userName,
         bool enforcingConnectionQuotas,
         uint16_t connectionUserQuota,
-        boost::shared_ptr<const AclData::bwHostRuleSet> globalBWRules,
-        boost::shared_ptr<const AclData::bwHostRuleSet> userBWRules)
+        boost::shared_ptr<AclData> localdata)
 {
     const std::string& hostName(getClientHost(connection.getMgmtId()));
 
@@ -227,74 +227,17 @@ bool ConnectionCounter::approveConnection(
                              C_OPENED, false, false);
 
     // Run global black/white list check
-    //
-    // TODO: The global check could be run way back in AsynchIO where
-    // disapproval would mean that the socket is not accepted. Or
-    // it may be accepted and closed right away without running any
-    // protocol and creating the connection churn that gets here.
-    //
     sys::SocketAddress sa(hostName, "");
+    bool okByHostList(true);
+    std::string hostLimitText;
     if (sa.isIp()) {
-        if (boost::shared_ptr<const AclData::bwHostRuleSet>() != globalBWRules) {
-            AclData::bwHostRuleSet::const_iterator it;
-            for (it=globalBWRules->begin(); it!=globalBWRules->end(); it++) {
-                if (it->getAclHost().match(hostName)) {
-                    // This host matches a global spec and controls the
-                    // allow/deny decision for this connection.
-                    AclResult res = it->getAclResult();
-                    if (res == DENY || res == DENYLOG) {
-                        // The result is deny
-                        QPID_LOG(trace, "ACL ConnectionApprover global rule " << it->toString()
-                                << " denies connection for host " << hostName << ", user "
-                                << userName);
-                        acl.reportConnectLimit(userName, hostName);
-                        return false;
-                    } else {
-                        // The result is allow
-                        QPID_LOG(trace, "ACL ConnectionApprover global rule " << it->toString()
-                                << " allows connection for host " << hostName << ", user "
-                                << userName);
-                        break;
-                    }
-                } else {
-                    // This rule in the global spec doesn't match and
-                    // does not control the allow/deny decision.
-                }
-            }
+        AclResult result = localdata->isAllowedConnection(userName, hostName, hostLimitText);
+        okByHostList = AclHelper::resultAllows(result);
+        if (okByHostList) {
+            QPID_LOG(trace, "ACL: ConnectionApprover host list " << hostLimitText);
         }
-
-        // Run user black/white list check
-        if (boost::shared_ptr<const AclData::bwHostRuleSet>() != userBWRules) {
-            AclData::bwHostRuleSet::const_iterator it;
-            for (it=userBWRules->begin(); it!=userBWRules->end(); it++) {
-                if (it->getAclHost().match(hostName)) {
-                    // This host matches a user spec and controls the
-                    // allow/deny decision for this connection.
-                    AclResult res = it->getAclResult();
-                    if (res == DENY || res == DENYLOG) {
-                        // The result is deny
-                        QPID_LOG(trace, "ACL ConnectionApprover user rule " << it->toString()
-                                << " denies connection for host " << hostName << ", user "
-                                << userName);
-                        acl.reportConnectLimit(userName, hostName);
-                        return false;
-                    } else {
-                        // The result is allow
-                        QPID_LOG(trace, "ACL ConnectionApprover user rule " << it->toString()
-                                << " allows connection for host " << hostName << ", user "
-                                << userName);
-                        break;
-                    }
-                } else {
-                    // This rule in the user's spec doesn't match and
-                    // does not control the allow/deny decision.
-                }
-            }
-        }
-    } else {
-        // Non-IP hosts don't get subjected to blacklist and whitelist
-        // checks.
     }
+
     // Approve total connections
     bool okTotal  = true;
     if (totalLimit > 0) {
@@ -313,6 +256,10 @@ bool ConnectionCounter::approveConnection(
                                       enforcingConnectionQuotas);
 
     // Emit separate log for each disapproval
+    if (!okByHostList) {
+        QPID_LOG(error, "ACL: ConnectionApprover host list " << hostLimitText
+                 << " Connection refused.");
+    }
     if (!okTotal) {
         QPID_LOG(error, "Client max total connection count limit of " << totalLimit
                  << " exceeded by '"
@@ -333,7 +280,7 @@ bool ConnectionCounter::approveConnection(
     }
 
     // Count/Event once for each disapproval
-    bool result = okTotal && okByIP && okByUser;
+    bool result = okByHostList && okTotal && okByIP && okByUser;
     if (!result) {
         acl.reportConnectLimit(userName, hostName);
     }
diff --git a/qpid/cpp/src/qpid/acl/AclConnectionCounter.h b/qpid/cpp/src/qpid/acl/AclConnectionCounter.h
index 6b8f396867..3683b573ff 100644
--- a/qpid/cpp/src/qpid/acl/AclConnectionCounter.h
+++ b/qpid/cpp/src/qpid/acl/AclConnectionCounter.h
@@ -97,8 +97,7 @@ public:
                            const std::string& userName,
                            bool enforcingConnectionQuotas,
                            uint16_t connectionLimit,
-                           boost::shared_ptr<const AclData::bwHostRuleSet> globalBWRules,
-                           boost::shared_ptr<const AclData::bwHostRuleSet> userBWRules
+                           boost::shared_ptr<AclData> localdata
                           );
 };
 
diff --git a/qpid/cpp/src/qpid/acl/AclData.cpp b/qpid/cpp/src/qpid/acl/AclData.cpp
index d325442353..a629e44d60 100644
--- a/qpid/cpp/src/qpid/acl/AclData.cpp
+++ b/qpid/cpp/src/qpid/acl/AclData.cpp
@@ -17,10 +17,13 @@
  */
 
 #include "qpid/acl/AclData.h"
+#include "qpid/acl/AclValidator.h"
 #include "qpid/log/Statement.h"
 #include "qpid/sys/IntegerTypes.h"
 #include <boost/lexical_cast.hpp>
 #include <boost/shared_ptr.hpp>
+#include <sstream>
+#include <iomanip>
 
 namespace qpid {
 namespace acl {
@@ -49,10 +52,11 @@ AclData::AclData():
     decisionMode(qpid::acl::DENY),
     transferAcl(false),
     aclSource("UNKNOWN"),
+    connectionDecisionMode(qpid::acl::ALLOW),
     connQuotaRuleSettings(new quotaRuleSet),
     queueQuotaRuleSettings(new quotaRuleSet),
     connBWHostsGlobalRules(new bwHostRuleSet),
-    connBWHostsRuleSettings(new bwHostUserRuleMap)
+    connBWHostsUserRules(new bwHostUserRuleMap)
 {
     for (unsigned int cnt=0; cnt< qpid::acl::ACTIONSIZE; cnt++) {
         actionList[cnt]=0;
@@ -74,12 +78,56 @@ void AclData::clear ()
         delete[] actionList[cnt];
     }
     transferAcl = false;
+    connectionDecisionMode = qpid::acl::ALLOW;
     connQuotaRuleSettings->clear();
     queueQuotaRuleSettings->clear();
     connBWHostsGlobalRules->clear();
-    connBWHostsRuleSettings->clear();
+    connBWHostsUserRules->clear();
 }
 
+void AclData::printDecisionRules(int userFieldWidth) {
+    AclValidator validator;
+    QPID_LOG(trace, "ACL: Decision rule cross reference");
+    for (int act=0; act<acl::ACTIONSIZE; act++) {
+        acl::Action action = acl::Action(act);
+        for (int obj=0; obj<acl::OBJECTSIZE; obj++) {
+            acl::ObjectType object = acl::ObjectType(obj);
+            if (actionList[act] != NULL && actionList[act][obj] != NULL) {
+                for (actObjItr aoitr = actionList[act][obj]->begin();
+                    aoitr != actionList[act][obj]->end();
+                    aoitr++) {
+                    std::string user = (*aoitr).first;
+                    ruleSetItr rsitr = (*aoitr).second.end();
+                    for (size_t rCnt=0; rCnt < (*aoitr).second.size(); rCnt++) {
+                        rsitr--;
+                        std::vector<int> candidates;
+                        validator.findPossibleLookupMatch(
+                            action, object, rsitr->props, candidates);
+                        std::stringstream ss;
+                        std::string sep("");
+                        for (std::vector<int>::const_iterator
+                            itr = candidates.begin(); itr != candidates.end(); itr++) {
+                            ss << sep << *itr;
+                            sep = ",";
+                        }
+                        QPID_LOG(trace, "ACL: User: "
+                            << std::setfill(' ') << std::setw(userFieldWidth +1) << std::left
+                            << user << " "
+                            << std::setfill(' ') << std::setw(acl::ACTION_STR_WIDTH +1) << std::left
+                            << AclHelper::getActionStr(action)
+                            << std::setfill(' ') << std::setw(acl::OBJECTTYPE_STR_WIDTH) << std::left
+                            << AclHelper::getObjectTypeStr(object)
+                            << " Rule: "
+                            << rsitr->toString() << " may match Lookups : ("
+                            << ss.str() << ")");
+                    }
+                }
+            } else {
+                // no rules for action/object
+            }
+        }
+    }
+}
 
 //
 // matchProp
@@ -619,22 +667,66 @@ void AclData::setConnGlobalRules (boost::shared_ptr<bwHostRuleSet> cgr) {
 }
 
 void AclData::setConnUserRules (boost::shared_ptr<bwHostUserRuleMap> hurm) {
-    connBWHostsRuleSettings = hurm;
+    connBWHostsUserRules = hurm;
 }
 
+AclResult AclData::isAllowedConnection(const std::string& userName,
+                                       const std::string& hostName,
+                                       std::string& logText) {
+    bool decisionMade(false);
+    AclResult result(ALLOW);
+    for (bwHostRuleSetItr it=connBWHostsGlobalRules->begin();
+                          it!=connBWHostsGlobalRules->end(); it++) {
+        if (it->getAclHost().match(hostName)) {
+            // This host matches a global spec and controls the
+            // allow/deny decision for this connection.
+            result = it->getAclResult();
+            logText = QPID_MSG("global rule " << it->toString()
+                    << (AclHelper::resultAllows(result) ? " allows" : " denies")
+                    << " connection for host " << hostName << ", user "
+                    << userName);
+            decisionMade = true;
+            break;
+        } else {
+            // This rule in the global spec doesn't match and
+            // does not control the allow/deny decision.
+        }
+    }
 
-//
-// Get user-specific black/white connection rule list
-//
-boost::shared_ptr<const AclData::bwHostRuleSet> AclData::getUserConnectionRules(const std::string& name){
-    AclData::bwHostUserRuleMapItr itrRule = connBWHostsRuleSettings->find(name);
-    if (itrRule == connBWHostsRuleSettings->end()) {
-        return boost::shared_ptr<const bwHostRuleSet>();
-    } else {
-        return boost::shared_ptr<const bwHostRuleSet>(&itrRule->second);
+    // Run user black/white list check
+    if (!decisionMade) {
+        bwHostUserRuleMapItr itrRule = connBWHostsUserRules->find(userName);
+        if (itrRule != connBWHostsUserRules->end()) {
+            for (bwHostRuleSetItr it=(*itrRule).second.begin();
+                 it!=(*itrRule).second.end(); it++) {
+                if (it->getAclHost().match(hostName)) {
+                    // This host matches a user spec and controls the
+                    // allow/deny decision for this connection.
+                    result = it->getAclResult();
+                    logText = QPID_MSG("global rule " << it->toString()
+                            << (AclHelper::resultAllows(result) ? " allows" : " denies")
+                            << " connection for host " << hostName << ", user "
+                            << userName);
+                    decisionMade = true;
+                    break;
+                } else {
+                    // This rule in the user's spec doesn't match and
+                    // does not control the allow/deny decision.
+                }
+            }
+        }
     }
-}
 
+    // Apply global connection mode
+    if (!decisionMade) {
+        result = connectionDecisionMode;
+        logText = QPID_MSG("default connection policy "
+                << (AclHelper::resultAllows(result) ? "allows" : "denies")
+                << " connection for host " << hostName << ", user "
+                << userName);
+    }
+    return result;
+}
 
 //
 //
diff --git a/qpid/cpp/src/qpid/acl/AclData.h b/qpid/cpp/src/qpid/acl/AclData.h
index 2167af66b8..105a5d9c67 100644
--- a/qpid/cpp/src/qpid/acl/AclData.h
+++ b/qpid/cpp/src/qpid/acl/AclData.h
@@ -69,7 +69,7 @@ public:
     typedef specPropertyMap::const_iterator                specPropertyMapItr;
 
     //
-    // rule
+    // Rule
     //
     // Created by AclReader and stored in a ruleSet vector for subsequent
     //  run-time lookup matching and allow/deny decisions.
@@ -92,6 +92,8 @@ public:
         bool                  pubExchNameMatchesBlank;
         std::string           pubExchName;
         std::vector<bool>     ruleHasUserSub;
+        std::string           lookupSource;
+        std::string           lookupHelp;
 
         Rule (int ruleNum, qpid::acl::AclResult res, specPropertyMap& p) :
             rawRuleNum(ruleNum),
@@ -106,6 +108,24 @@ public:
             ruleHasUserSub(PROPERTYSIZE, false)
             {}
 
+        // Variation of Rule for tracking PropertyDefs
+        // for AclValidation.
+        Rule (int ruleNum, qpid::acl::AclResult res, specPropertyMap& p,
+              const std::string& ls, const std::string& lh
+        ) :
+            rawRuleNum(ruleNum),
+            ruleMode(res),
+            props(p),
+            pubRoutingKeyInRule(false),
+            pubRoutingKey(),
+            pubExchNameInRule(false),
+            pubExchNameMatchesBlank(false),
+            pubExchName(),
+            ruleHasUserSub(PROPERTYSIZE, false),
+            lookupSource(ls),
+            lookupHelp(lh)
+            {}
+
 
         std::string toString () const {
             std::ostringstream ruleStr;
@@ -148,11 +168,12 @@ public:
     typedef  std::map<std::string, bwHostRuleSet> bwHostUserRuleMap; //<username, hosts-vector>
     typedef  bwHostUserRuleMap::const_iterator    bwHostUserRuleMapItr;
 
-    // Action*[] -> Object*[] -> map<user -> set<Rule> >
+    // Action*[] -> Object*[] -> map<user, set<Rule> >
     aclAction*           actionList[qpid::acl::ACTIONSIZE];
     qpid::acl::AclResult decisionMode;  // allow/deny[-log] if no matching rule found
     bool                 transferAcl;
     std::string          aclSource;
+    qpid::acl::AclResult connectionDecisionMode;
 
     AclResult lookup(
         const std::string&               id,        // actor id
@@ -172,10 +193,14 @@ public:
         return connBWHostsGlobalRules;
     }
 
-    boost::shared_ptr<const AclData::bwHostRuleSet> getUserConnectionRules(const std::string& name);
+    boost::shared_ptr<const bwHostUserRuleMap> getUserConnectionRules() {
+        return connBWHostsUserRules;
+    }
 
     bool matchProp(const std::string & src, const std::string& src1);
     void clear ();
+    void printDecisionRules(int userFieldWidth);
+
     static const std::string ACL_KEYWORD_USER_SUBST;
     static const std::string ACL_KEYWORD_DOMAIN_SUBST;
     static const std::string ACL_KEYWORD_USERDOMAIN_SUBST;
@@ -243,6 +268,19 @@ public:
         return "65530";
     }
 
+    /**
+     * isAllowedConnection
+     * Return true if this user is allowed to connect to this host.
+     * Return log text describing both success and failure.
+     */
+    AclResult isAllowedConnection(const std::string& userName,
+                                  const std::string& hostName,
+                                  std::string& logText);
+
+    AclResult connectionMode() const {
+        return connectionDecisionMode;
+    }
+
     AclData();
     virtual ~AclData();
 
@@ -277,7 +315,7 @@ private:
     boost::shared_ptr<bwHostRuleSet> connBWHostsGlobalRules;
 
     // Per-user host connection black/white rule set map
-    boost::shared_ptr<bwHostUserRuleMap> connBWHostsRuleSettings;
+    boost::shared_ptr<bwHostUserRuleMap> connBWHostsUserRules;
 };
 
 }} // namespace qpid::acl
diff --git a/qpid/cpp/src/qpid/acl/AclLexer.cpp b/qpid/cpp/src/qpid/acl/AclLexer.cpp
index d0f149db84..4006e5271f 100644
--- a/qpid/cpp/src/qpid/acl/AclLexer.cpp
+++ b/qpid/cpp/src/qpid/acl/AclLexer.cpp
@@ -32,7 +32,7 @@ namespace acl {
 
 // ObjectType
 const std::string objectNames[OBJECTSIZE] = {
-    "queue", "exchange", "broker", "link", "method", "query", "connection" };
+    "broker", "connection", "exchange", "link", "method", "query", "queue" };
 
 ObjectType AclHelper::getObjectType(const std::string& str) {
     for (int i=0; i< OBJECTSIZE; ++i) {
@@ -48,9 +48,9 @@ const std::string& AclHelper::getObjectTypeStr(const ObjectType o) {
 
 // Action
 const std::string actionNames[ACTIONSIZE] = {
-    "consume", "publish", "create", "access", "bind",
-    "unbind", "delete", "purge", "update", "move",
-    "redirect", "reroute" };
+    "access", "bind", "consume", "create", "delete",
+    "move", "publish", "purge", "redirect", "reroute",
+    "unbind", "update" };
 
 Action AclHelper::getAction(const std::string& str) {
     for (int i=0; i< ACTIONSIZE; ++i) {
@@ -133,4 +133,9 @@ const std::string& AclHelper::getAclResultStr(const AclResult r) {
     return resultNames[r];
 }
 
+bool AclHelper::resultAllows(const AclResult r) {
+    bool answer = r == ALLOW || r == ALLOWLOG;
+    return answer;
+}
+
 }} // namespace qpid::acl
diff --git a/qpid/cpp/src/qpid/acl/AclLexer.h b/qpid/cpp/src/qpid/acl/AclLexer.h
index a8032ddee7..d3df411afd 100644
--- a/qpid/cpp/src/qpid/acl/AclLexer.h
+++ b/qpid/cpp/src/qpid/acl/AclLexer.h
@@ -43,31 +43,35 @@ namespace acl {
 
     // ObjectType  shared between ACL spec and ACL authorise interface
     enum ObjectType {
-        OBJ_QUEUE,
-        OBJ_EXCHANGE,
         OBJ_BROKER,
+        OBJ_CONNECTION,
+        OBJ_EXCHANGE,
         OBJ_LINK,
         OBJ_METHOD,
         OBJ_QUERY,
-        OBJ_CONNECTION,
+        OBJ_QUEUE,
         OBJECTSIZE }; // OBJECTSIZE must be last in list
 
+    const int OBJECTTYPE_STR_WIDTH = 10;
+
     // Action  shared between ACL spec and ACL authorise interface
     enum Action {
-        ACT_CONSUME,
-        ACT_PUBLISH,
-        ACT_CREATE,
         ACT_ACCESS,
         ACT_BIND,
-        ACT_UNBIND,
+        ACT_CONSUME,
+        ACT_CREATE,
         ACT_DELETE,
-        ACT_PURGE,
-        ACT_UPDATE,
         ACT_MOVE,
+        ACT_PUBLISH,
+        ACT_PURGE,
         ACT_REDIRECT,
         ACT_REROUTE,
+        ACT_UNBIND,
+        ACT_UPDATE,
         ACTIONSIZE }; // ACTIONSIZE must be last in list
 
+    const int ACTION_STR_WIDTH = 8;
+
     // Property used in ACL authorize interface
     enum Property {
         PROP_NAME,
@@ -153,6 +157,7 @@ namespace acl {
         static QPID_BROKER_EXTERN const std::string&   getPropertyStr(const SpecProperty p);
         static QPID_BROKER_EXTERN AclResult              getAclResult(const std::string& str);
         static QPID_BROKER_EXTERN const std::string&  getAclResultStr(const AclResult r);
+        static QPID_BROKER_EXTERN bool                   resultAllows(const AclResult r);
 
         typedef std::set<Property>                  propSet;
         typedef boost::shared_ptr<propSet>          propSetPtr;
@@ -160,9 +165,6 @@ namespace acl {
         typedef std::map<Action, propSetPtr>        actionMap;
         typedef boost::shared_ptr<actionMap>        actionMapPtr;
         typedef std::pair<ObjectType, actionMapPtr> objectPair;
-        typedef std::map<ObjectType, actionMapPtr>  objectMap;
-        typedef objectMap::const_iterator           omCitr;
-        typedef boost::shared_ptr<objectMap>        objectMapPtr;
         typedef std::map<Property, std::string>     propMap;
         typedef propMap::const_iterator             propMapItr;
         typedef std::map<SpecProperty, std::string> specPropMap;
diff --git a/qpid/cpp/src/qpid/acl/AclReader.cpp b/qpid/cpp/src/qpid/acl/AclReader.cpp
index a0b4c67bf3..e8223c3570 100644
--- a/qpid/cpp/src/qpid/acl/AclReader.cpp
+++ b/qpid/cpp/src/qpid/acl/AclReader.cpp
@@ -26,6 +26,7 @@
 #include "qpid/log/Statement.h"
 #include "qpid/Exception.h"
 #include <boost/lexical_cast.hpp>
+#include <algorithm>
 
 #include <iomanip> // degug
 #include <iostream> // debug
@@ -55,11 +56,6 @@ namespace acl {
         return props.insert(propNvPair(p, v)).second;
     }
 
-    bool AclReader::aclRule::validate(const AclHelper::objectMapPtr& /*validationMap*/) {
-        // TODO - invalid rules won't ever be called in real life...
-        return true;
-    }
-
     // Debug aid
     std::string AclReader::aclRule::toString() {
         std::ostringstream oss;
@@ -89,6 +85,7 @@ namespace acl {
         d->clear();
         QPID_LOG(debug, "ACL: Load Rules");
         bool foundmode = false;
+        bool foundConnectionMode = false;
 
         rlCitr i = rules.end();
         for (int cnt = rules.size(); cnt; cnt--) {
@@ -96,6 +93,18 @@ namespace acl {
             QPID_LOG(debug, "ACL: Processing " << std::setfill(' ') << std::setw(2)
                 << cnt << " " << (*i)->toString());
 
+            if (!(*i)->actionAll && (*i)->objStatus == aclRule::VALUE &&
+                !validator.validateAllowedProperties(
+                    (*i)->action, (*i)->object, (*i)->props, false)) {
+                    // specific object/action has bad property
+                    // this rule gets ignored
+                    continue;
+            } else {
+                // action=all or object=none/all means the rule gets propagated
+                // possibly to many places.
+                // Invalid rule combinations are not propagated.
+            }
+
             if (!foundmode && (*i)->actionAll && (*i)->names.size() == 1
                 && (*((*i)->names.begin())).compare(AclData::ACL_KEYWORD_WILDCARD) == 0) {
                     d->decisionMode = (*i)->res;
@@ -105,18 +114,33 @@ namespace acl {
             } else if ((*i)->action == acl::ACT_CREATE && (*i)->object == acl::OBJ_CONNECTION) {
                 // Intercept CREATE CONNECTION rules process them into separate lists to
                 // be consumed in the connection approval code path.
+                propMap::const_iterator pName = (*i)->props.find(SPECPROP_NAME);
+                if (pName != (*i)->props.end()) {
+                    throw Exception(QPID_MSG("ACL: CREATE CONNECTION rule " << cnt << " must not have a 'name' property"));
+                }
                 propMap::const_iterator pHost = (*i)->props.find(SPECPROP_HOST);
                 if (pHost == (*i)->props.end()) {
                     throw Exception(QPID_MSG("ACL: CREATE CONNECTION rule " << cnt << " has no 'host' property"));
                 }
                 // create the connection rule
-                AclBWHostRule bwRule((*i)->res,
-                                     (pHost->second.compare(AclData::ACL_KEYWORD_ALL) != 0 ? pHost->second : ""));
+                bool allUsers = (*(*i)->names.begin()).compare(AclData::ACL_KEYWORD_WILDCARD) == 0;
+                bool allHosts = pHost->second.compare(AclData::ACL_KEYWORD_ALL) == 0;
+                AclBWHostRule bwRule((*i)->res, (allHosts ? "" : pHost->second));
 
                 // apply the rule globally or to user list
-                if ((*(*i)->names.begin()).compare(AclData::ACL_KEYWORD_WILDCARD) == 0) {
-                    // Rules for user "all" go into the gloabl list
-                    globalHostRules->insert( globalHostRules->begin(), bwRule );
+                if (allUsers) {
+                    if (allHosts) {
+                        // allow one specification of allUsers,allHosts
+                        if (foundConnectionMode) {
+                            throw Exception(QPID_MSG("ACL: only one CREATE CONNECTION rule for user=all and host=all allowed"));
+                        }
+                        foundConnectionMode = true;
+                        d->connectionDecisionMode = (*i)->res;
+                        QPID_LOG(trace, "ACL: Found connection mode: " << AclHelper::getAclResultStr( (*i)->res ));
+                    } else {
+                        // Rules for allUsers but not allHosts go into the global list
+                        globalHostRules->insert( globalHostRules->begin(), bwRule );
+                    }
                 } else {
                     // other rules go into binned rule sets for each user
                     for (nsCitr itr  = (*i)->names.begin();
@@ -127,7 +151,6 @@ namespace acl {
                 }
             } else {
                 AclData::Rule rule(cnt, (*i)->res, (*i)->props);
-
                 // Record which properties have the user substitution string
                 for (pmCitr pItr=rule.props.begin(); pItr!=rule.props.end(); pItr++) {
                     if ((pItr->second.find(AclData::ACL_KEYWORD_USER_SUBST, 0)       != std::string::npos) ||
@@ -179,7 +202,6 @@ namespace acl {
                             d->actionList[acnt][j] = NULL;
                     }
 
-                    // TODO: optimize this loop to limit to valid options only!!
                     for (int ocnt = ((*i)->objStatus != aclRule::VALUE ? 0
                         : (*i)->object);
                         ocnt < acl::OBJECTSIZE;
@@ -199,20 +221,23 @@ namespace acl {
                         for (nsCitr itr  = (allNames ? names.begin() : (*i)->names.begin());
                                     itr != (allNames ? names.end()   : (*i)->names.end());
                                     itr++) {
-                            AclData::actObjItr itrRule =
-                                d->actionList[acnt][ocnt]->find(*itr);
-
-                            if (itrRule == d->actionList[acnt][ocnt]->end()) {
-                                AclData::ruleSet rSet;
-                                rSet.push_back(rule);
-                                d->actionList[acnt][ocnt]->insert
-                                    (make_pair(std::string(*itr), rSet));
+                            if (validator.validateAllowedProperties(acl::Action(acnt),
+                                                                    acl::ObjectType(ocnt),
+                                                                    (*i)->props,
+                                                                    false)) {
+                                AclData::actObjItr itrRule =
+                                    d->actionList[acnt][ocnt]->find(*itr);
+
+                                if (itrRule == d->actionList[acnt][ocnt]->end()) {
+                                    AclData::ruleSet rSet;
+                                    rSet.push_back(rule);
+                                    d->actionList[acnt][ocnt]->insert
+                                        (make_pair(std::string(*itr), rSet));
+                                } else {
+                                        itrRule->second.push_back(rule);
+                                }
                             } else {
-                                // TODO add code to check for dead rules
-                                // allow peter create queue name=tmp <-- dead rule!!
-                                // allow peter create queue
-
-                                itrRule->second.push_back(rule);
+                                // Skip propagating this rule as it will never match.
                             }
                         }
                     }
@@ -241,7 +266,7 @@ namespace acl {
                     AclHelper::propertyMapToString(&rule.props)
                     << " for users {" <<
                     userstr.str().substr(0,userstr.str().length()-1)
-                    << "}" );
+                    << "}");
             }
         }
 
@@ -271,7 +296,6 @@ namespace acl {
 
     AclReader::AclReader(uint16_t theCliMaxConnPerUser, uint16_t theCliMaxQueuesPerUser) :
         lineNumber(0), contFlag(false),
-        validationMap(new AclHelper::objectMap),
         cliMaxConnPerUser (theCliMaxConnPerUser),
         connQuotaRulesExist(false),
         connQuota(new AclData::quotaRuleSet),
@@ -347,6 +371,8 @@ namespace acl {
         }
         printGlobalConnectRules();
         printUserConnectRules();
+        validator.tracePropertyDefs();
+        d->printDecisionRules( printNamesFieldWidth() );
 
         return 0;
     }
@@ -598,7 +624,9 @@ namespace acl {
         names.insert(name);
     }
 
-    // Debug aid
+    /**
+     * Emit debug logs exposing the name lists
+     */
     void AclReader::printNames() const {
         QPID_LOG(debug, "ACL: Group list: " << groups.size() << " groups found:" );
         std::string tmp("ACL: ");
@@ -622,6 +650,17 @@ namespace acl {
         QPID_LOG(debug, tmp);
     }
 
+    /**
+     * compute the width of longest user name
+     */
+    int AclReader::printNamesFieldWidth() const {
+        std::string::size_type max = 0;
+        for (nsCitr k=names.begin(); k!=names.end(); k++) {
+            max = std::max(max, (*k).length());
+        }
+        return max;
+    }
+
     bool AclReader::processAclLine(tokList& toks) {
         const unsigned toksSize = toks.size();
         if (toksSize < 4) {
@@ -709,12 +748,6 @@ namespace acl {
             }
         }
 
-        // If rule validates, add to rule list
-        if (!rule->validate(validationMap)) {
-            errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber
-                << ", Invalid object/action/property combination.";
-            return false;
-        }
         rules.push_back(rule);
 
         return true;
@@ -726,6 +759,9 @@ namespace acl {
         int cnt = 1;
         for (rlCitr i=rules.begin(); i<rules.end(); i++,cnt++) {
             QPID_LOG(debug, "ACL:   " << std::setfill(' ') << std::setw(2) << cnt << " " << (*i)->toString());
+            if (!(*i)->actionAll && (*i)->objStatus == aclRule::VALUE) {
+                (void)validator.validateAllowedProperties((*i)->action, (*i)->object, (*i)->props, true);
+            }
         }
     }
 
diff --git a/qpid/cpp/src/qpid/acl/AclReader.h b/qpid/cpp/src/qpid/acl/AclReader.h
index 1f3cf6dca1..24237a82ce 100644
--- a/qpid/cpp/src/qpid/acl/AclReader.h
+++ b/qpid/cpp/src/qpid/acl/AclReader.h
@@ -30,6 +30,7 @@
 #include "qpid/acl/AclData.h"
 #include "qpid/acl/Acl.h"
 #include "qpid/broker/AclModule.h"
+#include "qpid/acl/AclValidator.h"
 
 namespace qpid {
 namespace acl {
@@ -70,7 +71,6 @@ class AclReader {
         void setObjectType(const ObjectType o);
         void setObjectTypeAll();
         bool addProperty(const SpecProperty p, const std::string v);
-        bool validate(const AclHelper::objectMapPtr& validationMap);
         std::string toString(); // debug aid
       private:
         void processName(const std::string& name, const groupMap& groups);
@@ -97,7 +97,7 @@ class AclReader {
     nameSet                 names;
     groupMap                groups;
     ruleList                rules;
-    AclHelper::objectMapPtr validationMap;
+    AclValidator            validator;
     std::ostringstream      errorStream;
 
   public:
@@ -108,7 +108,7 @@ class AclReader {
 
   private:
     bool processLine(char* line);
-    void loadDecisionData( boost::shared_ptr<AclData> d);
+    void loadDecisionData(boost::shared_ptr<AclData> d);
     int tokenize(char* line, tokList& toks);
 
     bool processGroupLine(tokList& toks, const bool cont);
@@ -116,6 +116,7 @@ class AclReader {
     void addName(const std::string& name, nameSetPtr groupNameSet);
     void addName(const std::string& name);
     void printNames() const; // debug aid
+    int  printNamesFieldWidth() const;
 
     bool processAclLine(tokList& toks);
     void printRules() const; // debug aid
diff --git a/qpid/cpp/src/qpid/acl/AclValidator.cpp b/qpid/cpp/src/qpid/acl/AclValidator.cpp
index 655e7942fe..32be06d9e1 100644
--- a/qpid/cpp/src/qpid/acl/AclValidator.cpp
+++ b/qpid/cpp/src/qpid/acl/AclValidator.cpp
@@ -18,6 +18,7 @@
 
 #include "qpid/acl/AclValidator.h"
 #include "qpid/acl/AclData.h"
+#include "qpid/acl/AclLexer.h"
 #include "qpid/Exception.h"
 #include "qpid/log/Statement.h"
 #include "qpid/sys/IntegerTypes.h"
@@ -26,6 +27,7 @@
 #include <boost/bind.hpp>
 #include <numeric>
 #include <sstream>
+#include <iomanip>
 
 namespace qpid {
 namespace acl {
@@ -78,7 +80,7 @@ namespace acl {
         return oss.str();
     }
 
-    AclValidator::AclValidator(){
+    AclValidator::AclValidator() : propertyIndex(1) {
         validators.insert(Validator(acl::SPECPROP_MAXQUEUESIZELOWERLIMIT,
                           boost::shared_ptr<PropertyType>(
                             new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
@@ -135,41 +137,111 @@ namespace acl {
 
         // Insert allowed action/object/property sets (generated manually 20140712)
 #define RP registerProperties
-        RP("Broker::getTimestampConfig",            ACT_ACCESS,  OBJ_BROKER);
-        RP("ExchangeHandlerImpl::query",            ACT_ACCESS,  OBJ_EXCHANGE);
-        RP("ExchangeHandlerImpl::bound",            ACT_ACCESS,  OBJ_EXCHANGE, "queuename routingkey");
-        RP("ExchangeHandlerImpl::declare",          ACT_ACCESS,  OBJ_EXCHANGE, "type alternate durable autodelete");
-        RP("Authorise::access",                     ACT_ACCESS,  OBJ_EXCHANGE, "type durable");
-        RP("Authorise::access",                     ACT_ACCESS,  OBJ_EXCHANGE);
-        RP("ManagementAgent::handleMethodRequest",  ACT_ACCESS,  OBJ_METHOD, "schemapackage schemaclass");
-        RP("ManagementAgent::authorizeAgentMessage",ACT_ACCESS,  OBJ_METHOD, "schemapackage schemaclass");
-        RP("ManagementAgent::handleGetQuery",       ACT_ACCESS,  OBJ_QUERY, "schemaclass");
-        RP("Broker::queryQueue",                    ACT_ACCESS,  OBJ_QUEUE);
-        RP("QueueHandlerImpl::query",               ACT_ACCESS,  OBJ_QUEUE);
-        RP("QueueHandlerImpl::declare",             ACT_ACCESS,  OBJ_QUEUE, "alternate durable exclusive autodelete policytype maxqueuecount maxqueuesize");
-        RP("Authorise::access",                     ACT_ACCESS,  OBJ_QUEUE, "alternate durable exclusive autodelete policytype maxqueuecount maxqueuesize");
-        RP("Authorise::access",                     ACT_ACCESS,  OBJ_QUEUE);
-        RP("Broker::bind",                          ACT_BIND,    OBJ_EXCHANGE, "queuename routingkey");
-        RP("Authorise::outgoing",                   ACT_BIND,    OBJ_EXCHANGE, "queuename routingkey");
-        RP("MessageHandlerImpl::subscribe",         ACT_CONSUME, OBJ_QUEUE);
-        RP("Authorise::outgoing",                   ACT_CONSUME, OBJ_QUEUE);
-        RP("ConnectionHandler",                     ACT_CREATE,  OBJ_CONNECTION, "host");
-        RP("Broker::createQueue",                   ACT_CREATE,  OBJ_QUEUE, "alternate durable exclusive autodelete policytype paging maxpages maxpagefactor maxqueuecount maxqueuesize maxfilecount maxfilesize");
-        RP("Broker::createExchange",                ACT_CREATE,  OBJ_EXCHANGE, "type alternate durable autodelete");
-        RP("ConnectionHandler::Handler::open",      ACT_CREATE,  OBJ_LINK);
-        RP("Authorise::interlink",                  ACT_CREATE,  OBJ_LINK);
-        RP("Broker::deleteQueue",                   ACT_DELETE,  OBJ_QUEUE, "alternate durable exclusive autodelete policytype");
-        RP("Broker::deleteExchange",                ACT_DELETE,  OBJ_EXCHANGE, "type alternate durable");
-        RP("Broker::queueMoveMessages",             ACT_MOVE,    OBJ_QUEUE, "queuename");
-        RP("SemanticState::route",                  ACT_PUBLISH, OBJ_EXCHANGE, "routingkey");
-        RP("Authorise::incoming",                   ACT_PUBLISH, OBJ_EXCHANGE);
-        RP("Authorise::route",                      ACT_PUBLISH, OBJ_EXCHANGE, "routingkey");
-        RP("Queue::ManagementMethod",               ACT_PURGE,   OBJ_QUEUE);
-        RP("QueueHandlerImpl::purge",               ACT_PURGE,   OBJ_QUEUE);
-        RP("Broker::queueRedirect",                 ACT_REDIRECT,OBJ_QUEUE, "queuename");
-        RP("Queue::ManagementMethod",               ACT_REROUTE, OBJ_QUEUE, "exchangename");
-        RP("Broker::unbind",                        ACT_UNBIND,  OBJ_EXCHANGE, "queuename routingkey");
-        RP("Broker::setTimestampConfig",            ACT_UPDATE,  OBJ_BROKER);
+        RP( "Broker::getTimestampConfig",
+            "User querying message timestamp setting ",
+            ACT_ACCESS,  OBJ_BROKER);
+        RP( "ExchangeHandlerImpl::query",
+            "AMQP 0-10 protocol received 'query'     ",
+            ACT_ACCESS,  OBJ_EXCHANGE, "name");
+        RP( "ExchangeHandlerImpl::bound",
+            "AMQP 0-10 query binding                 ",
+            ACT_ACCESS,  OBJ_EXCHANGE, "name queuename routingkey");
+        RP( "ExchangeHandlerImpl::declare",
+            "AMQP 0-10 exchange declare              ",
+            ACT_ACCESS,  OBJ_EXCHANGE, "name type alternate durable autodelete");
+        RP( "Authorise::access",
+            "AMQP 1.0 exchange access                ",
+            ACT_ACCESS,  OBJ_EXCHANGE, "name type durable");
+        RP( "Authorise::access",
+            "AMQP 1.0 node resolution                ",
+            ACT_ACCESS,  OBJ_EXCHANGE, "name");
+        RP( "ManagementAgent::handleMethodRequest",
+            "Management method request               ",
+            ACT_ACCESS,  OBJ_METHOD, "name schemapackage schemaclass");
+        RP( "ManagementAgent::authorizeAgentMessage",
+            "Management agent method request         ",
+            ACT_ACCESS,  OBJ_METHOD, "name schemapackage schemaclass");
+        RP( "ManagementAgent::handleGetQuery",
+            "Management agent query                  ",
+            ACT_ACCESS,  OBJ_QUERY, "name schemaclass");
+        RP( "Broker::queryQueue",
+            "QMF 'query queue' method                ",
+            ACT_ACCESS,  OBJ_QUEUE, "name");
+        RP( "QueueHandlerImpl::query",
+            "AMQP 0-10 query                         ",
+            ACT_ACCESS,  OBJ_QUEUE, "name");
+        RP( "QueueHandlerImpl::declare",
+            "AMQP 0-10 queue declare                 ",
+            ACT_ACCESS,  OBJ_QUEUE, "name alternate durable exclusive autodelete policytype maxqueuecount maxqueuesize");
+        RP( "Authorise::access",
+            "AMQP 1.0 queue access                   ",
+            ACT_ACCESS,  OBJ_QUEUE, "name alternate durable exclusive autodelete policytype maxqueuecount maxqueuesize");
+        RP( "Authorise::access",
+            "AMQP 1.0 node resolution                ",
+            ACT_ACCESS,  OBJ_QUEUE, "name");
+        RP( "Broker::bind",
+            "AMQP 0-10 or QMF bind request           ",
+            ACT_BIND,    OBJ_EXCHANGE, "name queuename routingkey");
+        RP( "Authorise::outgoing",
+            "AMQP 1.0 new outgoing link from exchange",
+            ACT_BIND,    OBJ_EXCHANGE, "name queuename routingkey");
+        RP( "MessageHandlerImpl::subscribe",
+            "AMQP 0-10 subscribe request             ",
+            ACT_CONSUME, OBJ_QUEUE, "name");
+        RP( "Authorise::outgoing",
+            "AMQP 1.0 new outgoing link from queue   ",
+            ACT_CONSUME, OBJ_QUEUE, "name");
+        RP( "ConnectionHandler",
+            "TCP/IP connection creation              ",
+            ACT_CREATE,  OBJ_CONNECTION, "host");
+        RP( "Broker::createExchange",
+            "Create exchange                         ",
+            ACT_CREATE,  OBJ_EXCHANGE, "name type alternate durable autodelete");
+        RP( "ConnectionHandler::Handler::open",
+            "Interbroker link creation               ",
+            ACT_CREATE,  OBJ_LINK);
+        RP( "Authorise::interlink",
+            "Interbroker link creation               ",
+            ACT_CREATE,  OBJ_LINK);
+        RP( "Broker::createQueue",
+            "Create queue                            ",
+            ACT_CREATE,  OBJ_QUEUE, "name alternate durable exclusive autodelete policytype paging maxpages maxpagefactor maxqueuecount maxqueuesize maxfilecount maxfilesize");
+        RP( "Broker::deleteExchange",
+            "Delete exchange                         ",
+            ACT_DELETE,  OBJ_EXCHANGE, "name type alternate durable");
+        RP( "Broker::deleteQueue",
+            "Delete queue                            ",
+            ACT_DELETE,  OBJ_QUEUE, "name alternate durable exclusive autodelete policytype");
+        RP( "Broker::queueMoveMessages",
+            "Management 'move queue' request         ",
+            ACT_MOVE,    OBJ_QUEUE, "name queuename");
+        RP( "SemanticState::route",
+            "AMQP 0-10 received message processing   ",
+            ACT_PUBLISH, OBJ_EXCHANGE, "name routingkey");
+        RP( "Authorise::incoming",
+            "AMQP 1.0 establish sender link to queue ",
+            ACT_PUBLISH, OBJ_EXCHANGE, "routingkey");
+        RP( "Authorise::route",
+            "AMQP 1.0 received message processing    ",
+            ACT_PUBLISH, OBJ_EXCHANGE, "name routingkey");
+        RP( "Queue::ManagementMethod",
+            "Management 'purge queue' request        ",
+            ACT_PURGE,   OBJ_QUEUE, "name");
+        RP( "QueueHandlerImpl::purge",
+            "Management 'purge queue' request        ",
+            ACT_PURGE,   OBJ_QUEUE, "name");
+        RP( "Broker::queueRedirect",
+            "Management 'redirect queue' request     ",
+            ACT_REDIRECT,OBJ_QUEUE, "name queuename");
+        RP( "Queue::ManagementMethod",
+            "Management 'reroute queue' request      ",
+            ACT_REROUTE, OBJ_QUEUE, "name exchangename");
+        RP( "Broker::unbind",
+            "Management 'unbind exchange' request    ",
+            ACT_UNBIND,  OBJ_EXCHANGE, "name queuename routingkey");
+        RP( "Broker::setTimestampConfig",
+            "User modifying message timestamp setting",
+            ACT_UPDATE,  OBJ_BROKER);
     }
 
     AclValidator::~AclValidator(){
@@ -189,10 +261,10 @@ namespace acl {
                         std::for_each(d->actionList[cnt][cnt1]->begin(),
                                       d->actionList[cnt][cnt1]->end(),
                                       boost::bind(&AclValidator::validateRuleSet, this, _1));
-                    }//if
-                }//for
-            }//if
-        }//for
+                    }
+                }
+            }
+        }
     }
 
     void AclValidator::validateRuleSet(std::pair<const std::string, qpid::acl::AclData::ruleSet>& rules){
@@ -225,23 +297,155 @@ namespace acl {
     }
 
     /**
+     * validateAllowedProperties
+     * verify that at least one lookup definition can satisfy this
+     * action/object/props tuple.
+     * Return false and conditionally emit a warning log entry if the
+     * incoming definition can not be matched.
+     */
+    bool AclValidator::validateAllowedProperties(qpid::acl::Action action,
+                                                 qpid::acl::ObjectType object,
+                                                 const AclData::specPropertyMap& props,
+                                                 bool emitLog) const {
+        // No rules defined means no match
+        if (!allowedSpecProperties[action][object].get()) {
+            if (emitLog) {
+                QPID_LOG(warning, "ACL rule ignored: Broker never checks for rules with action: '"
+                    << AclHelper::getActionStr(action) << "' and object: '"
+                    << AclHelper::getObjectTypeStr(object) << "'");
+            }
+            return false;
+        }
+        // two empty property sets is a match
+        if (allowedSpecProperties[action][object]->size() == 0) {
+            if ((props.size() == 0) ||
+                (props.size() == 1 && props.find(acl::SPECPROP_NAME) != props.end())) {
+                return true;
+            }
+        }
+        // Scan vector of rules looking for one that matches all properties
+        bool validRuleFound = false;
+        for (std::vector<AclData::Rule>::const_iterator
+            ruleItr  = allowedSpecProperties[action][object]->begin();
+            ruleItr != allowedSpecProperties[action][object]->end() && !validRuleFound;
+            ruleItr++) {
+            // Scan one rule
+            validRuleFound = true;
+            for(AclData::specPropertyMapItr itr = props.begin();
+                itr != props.end();
+                itr++) {
+                if ((*itr).first != acl::SPECPROP_NAME &&
+                    ruleItr->props.find((*itr).first) ==
+                    ruleItr->props.end()) {
+                    // Test property not found in this rule
+                    validRuleFound = false;
+                    break;
+                }
+            }
+        }
+        if (!validRuleFound) {
+            if (emitLog) {
+                QPID_LOG(warning, "ACL rule ignored: Broker checks for rules with action: '"
+                    << AclHelper::getActionStr(action) << "' and object: '"
+                    << AclHelper::getObjectTypeStr(object)
+                    << "' but will never match with property set: "
+                    << AclHelper::propertyMapToString(&props));
+            }
+            return false;
+        }
+        return true;
+    }
+
+    /**
+     * Return a list of indexes of definitions that this lookup might match
+     */
+    void AclValidator::findPossibleLookupMatch(qpid::acl::Action action,
+                                               qpid::acl::ObjectType object,
+                                               const AclData::specPropertyMap& props,
+                                               std::vector<int>& result) const {
+        if (!allowedSpecProperties[action][object].get()) {
+            return;
+        } else {
+            // Scan vector of rules returning the indexes of all that match
+            bool validRuleFound;
+            for (std::vector<AclData::Rule>::const_iterator
+                ruleItr  = allowedSpecProperties[action][object]->begin();
+                ruleItr != allowedSpecProperties[action][object]->end();
+                ruleItr++) {
+                // Scan one rule
+                validRuleFound = true;
+                for(AclData::specPropertyMapItr
+                    itr = props.begin(); itr != props.end(); itr++) {
+                    if ((*itr).first != acl::SPECPROP_NAME &&
+                        ruleItr->props.find((*itr).first) ==
+                        ruleItr->props.end()) {
+                        // Test property not found in this rule
+                        validRuleFound = false;
+                        break;
+                    }
+                }
+                if (validRuleFound) {
+                    result.push_back(ruleItr->rawRuleNum);
+                }
+            }
+        }
+        return;
+    }
+
+    /**
+     * Emit trace log of original property definitions
+     */
+    void AclValidator::tracePropertyDefs() {
+        QPID_LOG(trace, "ACL: Definitions of action, object, (allowed properties) lookups");
+        for (int iA=0; iA<acl::ACTIONSIZE; iA++) {
+            for (int iO=0; iO<acl::OBJECTSIZE; iO++) {
+                if (allowedSpecProperties[iA][iO].get()) {
+                    for (std::vector<AclData::Rule>::const_iterator
+                        ruleItr  = allowedSpecProperties[iA][iO]->begin();
+                        ruleItr != allowedSpecProperties[iA][iO]->end();
+                        ruleItr++) {
+                        std::string pstr;
+                        for (AclData::specPropertyMapItr pMItr  = ruleItr->props.begin();
+                            pMItr != ruleItr->props.end();
+                            pMItr++) {
+                            pstr += AclHelper::getPropertyStr((SpecProperty) pMItr-> first);
+                            pstr += ",";
+                        }
+                        QPID_LOG(trace, "ACL: Lookup "
+                            << std::setfill(' ') << std::setw(2)
+                            << ruleItr->rawRuleNum << ": "
+                            << ruleItr->lookupHelp << " "
+                            << std::setfill(' ') << std::setw(acl::ACTION_STR_WIDTH +1) << std::left
+                            << AclHelper::getActionStr(acl::Action(iA))
+                            << std::setfill(' ') << std::setw(acl::OBJECTTYPE_STR_WIDTH) << std::left
+                            << AclHelper::getObjectTypeStr(acl::ObjectType(iO))
+                            << " (" << pstr.substr(0, pstr.length()-1) << ")");
+                    }
+                }
+            }
+        }
+    }
+
+    /**
      * Construct a record of all the calls that the broker will
      * make to acl::authorize and the properties for each call.
      * From that create the list of all the spec properties that
      * users are then allowed to specify in acl rule files.
      */
     void AclValidator::registerProperties(
-        const std::string& /* source */,
+        const std::string& source,
+        const std::string& description,
         Action action,
         ObjectType object,
         const std::string& properties) {
         if (!allowedProperties[action][object].get()) {
             boost::shared_ptr<std::set<Property> > t1(new std::set<Property>());
             allowedProperties[action][object] = t1;
-            boost::shared_ptr<std::set<SpecProperty> > t2(new std::set<SpecProperty>());
+            boost::shared_ptr<std::vector<AclData::Rule> > t2(new std::vector<AclData::Rule>());
             allowedSpecProperties[action][object] = t2;
         }
         std::vector<std::string> props = split(properties, " ");
+        AclData::specPropertyMap spm;
         for (size_t i=0; i<props.size(); i++) {
             Property prop = AclHelper::getProperty(props[i]);
             allowedProperties[action][object]->insert(prop);
@@ -250,35 +454,39 @@ namespace acl {
             switch (prop) {
                 // Cases where broker supplies a property but Acl has upper/lower limit for it
                 case PROP_MAXPAGES:
-                    allowedSpecProperties[action][object]->insert(SPECPROP_MAXPAGESLOWERLIMIT);
-                    allowedSpecProperties[action][object]->insert(SPECPROP_MAXPAGESUPPERLIMIT);
+                    spm[SPECPROP_MAXPAGESLOWERLIMIT]="";
+                    spm[SPECPROP_MAXPAGESUPPERLIMIT]="";
                     break;
                 case PROP_MAXPAGEFACTOR:
-                    allowedSpecProperties[action][object]->insert(SPECPROP_MAXPAGEFACTORLOWERLIMIT);
-                    allowedSpecProperties[action][object]->insert(SPECPROP_MAXPAGEFACTORUPPERLIMIT);
+                    spm[SPECPROP_MAXPAGEFACTORLOWERLIMIT]="";
+                    spm[SPECPROP_MAXPAGEFACTORUPPERLIMIT]="";
                     break;
                 case PROP_MAXQUEUESIZE:
-                    allowedSpecProperties[action][object]->insert(SPECPROP_MAXQUEUESIZELOWERLIMIT);
-                    allowedSpecProperties[action][object]->insert(SPECPROP_MAXQUEUESIZEUPPERLIMIT);
+                    spm[SPECPROP_MAXQUEUESIZELOWERLIMIT]="";
+                    spm[SPECPROP_MAXQUEUESIZEUPPERLIMIT]="";
                     break;
                 case PROP_MAXQUEUECOUNT:
-                    allowedSpecProperties[action][object]->insert(SPECPROP_MAXQUEUECOUNTLOWERLIMIT);
-                    allowedSpecProperties[action][object]->insert(SPECPROP_MAXQUEUECOUNTUPPERLIMIT);
+                    spm[SPECPROP_MAXQUEUECOUNTLOWERLIMIT]="";
+                    spm[SPECPROP_MAXQUEUECOUNTUPPERLIMIT]="";
                     break;
                 case PROP_MAXFILESIZE:
-                    allowedSpecProperties[action][object]->insert(SPECPROP_MAXFILESIZELOWERLIMIT);
-                    allowedSpecProperties[action][object]->insert(SPECPROP_MAXFILESIZEUPPERLIMIT);
+                    spm[SPECPROP_MAXFILESIZELOWERLIMIT]="";
+                    spm[SPECPROP_MAXFILESIZEUPPERLIMIT]="";
                     break;
                 case PROP_MAXFILECOUNT:
-                    allowedSpecProperties[action][object]->insert(SPECPROP_MAXFILECOUNTLOWERLIMIT);
-                    allowedSpecProperties[action][object]->insert(SPECPROP_MAXFILECOUNTUPPERLIMIT);
+                    spm[SPECPROP_MAXFILECOUNTLOWERLIMIT]="";
+                    spm[SPECPROP_MAXFILECOUNTUPPERLIMIT]="";
                     break;
                 default:
                     // Cases where broker supplies a property and Acl matches it directly
-                    allowedSpecProperties[action][object]->insert( SpecProperty(prop) );
+                    SpecProperty sp = SpecProperty(prop);
+                    spm[ sp ]="";
                     break;
             }
         }
+        AclData::Rule someProps(propertyIndex, acl::ALLOW, spm, source, description);
+        propertyIndex++;
+        allowedSpecProperties[action][object]->push_back(someProps);
     }
 
 }}
diff --git a/qpid/cpp/src/qpid/acl/AclValidator.h b/qpid/cpp/src/qpid/acl/AclValidator.h
index 03a80c5b09..8f555797c2 100644
--- a/qpid/cpp/src/qpid/acl/AclValidator.h
+++ b/qpid/cpp/src/qpid/acl/AclValidator.h
@@ -24,6 +24,7 @@
 #include "qpid/acl/AclData.h"
 #include "qpid/sys/IntegerTypes.h"
 #include <boost/shared_ptr.hpp>
+#include <boost/concept_check.hpp>
 #include <vector>
 #include <sstream>
 
@@ -65,8 +66,8 @@ class AclValidator {
     typedef std::pair<acl::SpecProperty,boost::shared_ptr<PropertyType> > Validator;
     typedef std::map<acl::SpecProperty,boost::shared_ptr<PropertyType> > ValidatorMap;
     typedef ValidatorMap::iterator ValidatorItr;
-    typedef boost::shared_ptr<std::set<Property> >     AllowedProperties    [ACTIONSIZE][OBJECTSIZE];
-    typedef boost::shared_ptr<std::set<SpecProperty> > AllowedSpecProperties[ACTIONSIZE][OBJECTSIZE];
+    typedef boost::shared_ptr<std::set<Property> >         AllowedProperties    [ACTIONSIZE][OBJECTSIZE];
+    typedef boost::shared_ptr<std::vector<AclData::Rule> > AllowedSpecProperties[ACTIONSIZE][OBJECTSIZE];
 
     ValidatorMap validators;
     AllowedProperties     allowedProperties;
@@ -78,14 +79,26 @@ public:
     void validateRule(qpid::acl::AclData::Rule& rule);
     void validateProperty(std::pair<const qpid::acl::SpecProperty, std::string>& prop);
     void validate(boost::shared_ptr<AclData> d);
+    bool validateAllowedProperties(qpid::acl::Action action,
+                                   qpid::acl::ObjectType object,
+                                   const AclData::specPropertyMap& props,
+                                   bool emitLog) const;
+    void findPossibleLookupMatch(qpid::acl::Action action,
+                                 qpid::acl::ObjectType object,
+                                 const AclData::specPropertyMap& props,
+                                 std::vector<int>& result) const;
+    void tracePropertyDefs();
+
     AclValidator();
     ~AclValidator();
 
 private:
     void registerProperties(const std::string& source,
+                            const std::string& description,
                             Action action,
                             ObjectType object,
                             const std::string& properties = "");
+    int propertyIndex;
 };
 
 }} // namespace qpid::acl
diff --git a/qpid/cpp/src/qpid/broker/Broker.cpp b/qpid/cpp/src/qpid/broker/Broker.cpp
index b981a6d0fe..3afaf43b81 100644
--- a/qpid/cpp/src/qpid/broker/Broker.cpp
+++ b/qpid/cpp/src/qpid/broker/Broker.cpp
@@ -607,7 +607,7 @@ Manageable::status_t Broker::ManagementMethod (uint32_t methodId,
         _qmf::ArgsBrokerQueueMoveMessages& moveArgs=
             dynamic_cast<_qmf::ArgsBrokerQueueMoveMessages&>(args);
         QPID_LOG (debug, "Broker::queueMoveMessages()");
-        if (queueMoveMessages(moveArgs.i_srcQueue, moveArgs.i_destQueue, moveArgs.i_qty, 
+        if (queueMoveMessages(moveArgs.i_srcQueue, moveArgs.i_destQueue, moveArgs.i_qty,
                               moveArgs.i_filter, getCurrentPublisher()) >=0)
             status = Manageable::STATUS_OK;
         else
@@ -1245,7 +1245,7 @@ Manageable::status_t Broker::queueRedirect(const std::string& srcQueue,
             if (!acl->authorise((context)?context->getUserId():"", acl::ACT_REDIRECT, acl::OBJ_QUEUE, srcQ->getName(), &params))
                 throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied redirect request from " << ((context)?context->getUserId():"(uknown)")));
         }
-                                  
+
         queueRedirectDestroy(srcQ, tgtQ, true);
 
         return Manageable::STATUS_OK;
diff --git a/qpid/cpp/src/tests/Acl.cpp b/qpid/cpp/src/tests/Acl.cpp
index 75a52c8ca1..9c3de0de62 100644
--- a/qpid/cpp/src/tests/Acl.cpp
+++ b/qpid/cpp/src/tests/Acl.cpp
@@ -45,6 +45,13 @@ QPID_AUTO_TEST_CASE(TestLexerObjectEnums) {
     OBJ_ENUMS(OBJ_METHOD,     "method");
     OBJ_ENUMS(OBJ_QUERY,      "query");
     OBJ_ENUMS(OBJ_CONNECTION, "connection");
+    int maxLen = 0;
+    for (int i=0; i<acl::OBJECTSIZE; i++) {
+        int thisLen = AclHelper::getObjectTypeStr( ObjectType(i) ).length();
+        if (thisLen > maxLen)
+            maxLen = thisLen;
+    }
+    BOOST_CHECK_EQUAL(maxLen, acl::OBJECTTYPE_STR_WIDTH);
 }
 
 #define ACT_ENUMS(e, s) \
@@ -65,6 +72,13 @@ QPID_AUTO_TEST_CASE(TestLexerActionEnums) {
     ACT_ENUMS(ACT_MOVE,     "move");
     ACT_ENUMS(ACT_REDIRECT, "redirect");
     ACT_ENUMS(ACT_REROUTE,  "reroute");
+    int maxLen = 0;
+    for (int i=0; i<acl::ACTIONSIZE; i++) {
+        int thisLen = AclHelper::getActionStr( Action(i) ).length();
+        if (thisLen > maxLen)
+            maxLen = thisLen;
+    }
+    BOOST_CHECK_EQUAL(maxLen, acl::ACTION_STR_WIDTH);
 }
 
 #define PROP_ENUMS(e, s) \
diff --git a/qpid/cpp/src/tests/acl.py b/qpid/cpp/src/tests/acl.py
index 5f5d1e01fe..75aa39295a 100755
--- a/qpid/cpp/src/tests/acl.py
+++ b/qpid/cpp/src/tests/acl.py
@@ -46,10 +46,10 @@ class ACLTests(TestBase010):
         parms = {'username':user, 'password':passwd, 'sasl_mechanisms':'PLAIN'}
         brokerurl="%s:%s" %(self.broker.host, self.broker.port)
         connection = qpid.messaging.Connection(brokerurl, **parms)
-        connection.open()       
+        connection.open()
         return connection
 
-    # For connection limit tests this function 
+    # For connection limit tests this function
     #  throws if the connection won't start
     #  returns a connection that the caller can close if he likes.
     def get_connection(self, user, passwd):
@@ -2115,6 +2115,7 @@ class ACLTests(TestBase010):
         aclf.write('acl allow all access method\n')
         # this should let bob access the timestamp configuration
         aclf.write('acl allow bob@QPID access broker\n')
+        aclf.write('acl allow bob@QPID update broker\n')
         aclf.write('acl allow admin@QPID all all\n')
         aclf.write('acl deny all all')
         aclf.close()
@@ -2239,7 +2240,7 @@ class ACLTests(TestBase010):
             self.LookupPublish(u, "company.topic", "private.audit.This", "allow-log")
 
         for u in uInTest:
-            for a in action_all:
+            for a in ['bind', 'unbind', 'access', 'publish']:
                 self.Lookup(u, a, "exchange", "company.topic", {"routingkey":"private.audit.This"}, "allow-log")
 
         for u in uOutTest:
@@ -3709,6 +3710,179 @@ class ACLTests(TestBase010):
             self.assertEqual(403,e.args[0].error_code)
             self.fail("ACL should allow exchange delete request for edae3h");
 
+    #=====================================
+    # 'create connection' tests
+    #=====================================
+#    def test_connect_mode_file_rejects_two_defaults(self):
+#        """
+#        Should reject a file with two connect mode statements
+#        """
+#        aclf = self.get_acl_file()
+#        aclf.write('acl allow all create connection host=all\n')
+#        aclf.write('acl allow all create connection host=all\n')
+#        aclf.close()
+#
+#        result = self.reload_acl()
+#        if (result):
+#            pass
+#        else:
+#            self.fail(result)
+
+    def test_connect_mode_accepts_host_spec_formats(self):
+        """
+        Should accept host specs of various forms
+        """
+        aclf = self.get_acl_file()
+        aclf.write('acl allow bob@QPID create connection host=all\n')
+        aclf.write('acl allow bob@QPID create connection host=1.1.1.1\n')
+        aclf.write('acl allow bob@QPID create connection host=1.1.1.1,2.2.2.2\n')
+        aclf.write('acl allow bob@QPID create connection host=localhost\n')
+        aclf.write('acl allow all all\n')
+        aclf.close()
+
+        result = self.reload_acl()
+        if (result):
+            self.fail(result)
+
+    def test_connect_mode_allow_all_mode(self):
+        """
+        Should allow one 'all', 'all'
+        """
+        aclf = self.get_acl_file()
+        aclf.write('acl allow all create connection host=all\n')
+        aclf.write('acl allow all all\n')
+        aclf.close()
+
+        result = self.reload_acl()
+        if (result):
+            self.fail(result)
+
+        session = self.get_session('bob','bob')
+
+
+    def test_connect_mode_allow_all_localhost(self):
+        """
+        Should allow 'all' 'localhost'
+        """
+        aclf = self.get_acl_file()
+        aclf.write('acl allow all create connection host=localhost\n')
+        aclf.write('acl deny  all create connection host=all\n')
+        aclf.write('acl allow all all\n')
+        aclf.close()
+
+        result = self.reload_acl()
+        if (result):
+            self.fail(result)
+
+        session = self.get_session('bob','bob')
+
+
+    def test_connect_mode_global_deny(self):
+        """
+        Should allow 'all' 'localhost'
+        """
+        aclf = self.get_acl_file()
+        aclf.write('acl allow all create connection host=localhost\n')
+        aclf.write('acl deny  all create connection host=all\n')
+        aclf.write('acl allow all all\n')
+        aclf.close()
+
+        result = self.reload_acl()
+        if (result):
+            self.fail(result)
+
+        session = self.get_session('bob','bob')
+
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"127.0.0.1"}, "allow")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"127.0.0.2"}, "deny")
+
+
+    def test_connect_mode_global_range(self):
+        """
+        Should allow 'all' 'localhost'
+        """
+        aclf = self.get_acl_file()
+        aclf.write('acl allow all create connection host=10.0.0.0,10.255.255.255\n')
+        aclf.write('acl allow all create connection host=localhost\n')
+        aclf.write('acl deny  all create connection host=all\n')
+        aclf.write('acl allow all all\n')
+        aclf.close()
+
+        result = self.reload_acl()
+        if (result):
+            self.fail(result)
+
+        session = self.get_session('bob','bob')
+
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"0.0.0.0"},        "deny")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"9.255.255.255"},  "deny")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"10.0.0.0"},       "allow")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"10.255.255.255"}, "allow")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"11.0.0.0"},       "deny")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"255.255.255.255"},"deny")
+
+
+    def test_connect_mode_nested_ranges(self):
+        """
+        Tests nested ranges for single user
+        """
+        aclf = self.get_acl_file()
+        aclf.write('acl deny-log  bob@QPID create connection host=10.0.1.0,10.0.1.255\n')
+        aclf.write('acl allow-log bob@QPID create connection host=10.0.0.0,10.255.255.255\n')
+        aclf.write('acl deny-log  bob@QPID create connection host=all\n')
+        aclf.write('acl allow all create connection host=localhost\n')
+        aclf.write('acl allow all all\n')
+        aclf.close()
+
+        result = self.reload_acl()
+        if (result):
+            self.fail(result)
+
+        session = self.get_session('bob','bob')
+
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"0.0.0.0"},        "deny-log")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"9.255.255.255"},  "deny-log")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"10.0.0.0"},       "allow-log")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"10.0.0.255"},     "allow-log")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"10.0.1.0"},       "deny-log")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"10.0.1.255"},     "deny-log")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"10.0.2.0"},       "allow-log")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"10.255.255.255"}, "allow-log")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"11.0.0.0"},       "deny-log")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"255.255.255.255"},"deny-log")
+
+
+    def test_connect_mode_user_ranges(self):
+        """
+        Two user ranges should not interfere with each other
+        """
+        aclf = self.get_acl_file()
+        aclf.write('acl allow-log bob@QPID create connection host=10.0.0.0,10.255.255.255\n')
+        aclf.write('acl deny-log  bob@QPID create connection host=all\n')
+        aclf.write('acl allow-log cat@QPID create connection host=192.168.0.0,192.168.255.255\n')
+        aclf.write('acl deny-log  cat@QPID create connection host=all\n')
+        aclf.write('acl allow all create connection host=localhost\n')
+        aclf.write('acl allow all all\n')
+        aclf.close()
+
+        result = self.reload_acl()
+        if (result):
+            self.fail(result)
+
+        session = self.get_session('bob','bob')
+
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"0.0.0.0"},        "deny-log")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"9.255.255.255"},  "deny-log")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"10.0.0.0"},       "allow-log")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"10.255.255.255"}, "allow-log")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"11.0.0.0"},       "deny-log")
+        self.Lookup("bob@QPID", "create", "connection", "", {"host":"255.255.255.255"},"deny-log")
+        self.Lookup("cat@QPID", "create", "connection", "", {"host":"0.0.0.0"},        "deny-log")
+        self.Lookup("cat@QPID", "create", "connection", "", {"host":"192.167.255.255"},"deny-log")
+        self.Lookup("cat@QPID", "create", "connection", "", {"host":"192.168.0.0"},    "allow-log")
+        self.Lookup("cat@QPID", "create", "connection", "", {"host":"192.168.255.255"},"allow-log")
+        self.Lookup("cat@QPID", "create", "connection", "", {"host":"192.169.0.0"},    "deny-log")
+        self.Lookup("cat@QPID", "create", "connection", "", {"host":"255.255.255.255"},"deny-log")
 
 
 class BrokerAdmin: