QPID-3892 C++ broker add routing key wildcard support to Acl 'publish exchange' lookups. Although this patch does not address the original issue's regex request it provides the desired functionality in a more comprehensive manner.

* Acl publish exchange rules may specify routing keys using the topic exchange syntax with '*' and '#' wildcard match tokens. 
* Acl lookups hook in to the broker's topic exchange key match code to perform the wildcard match.
* Acl rules written using the old Acl wildcard syntax (with a single trailing '*') will continue to work the same as before.




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

6 files changed, 96 insertions, 57 deletions

diff --git a/cpp/src/qpid/acl/AclData.cpp b/cpp/src/qpid/acl/AclData.cpp
index da7f240a9b..a07176dc76 100644
--- a/cpp/src/qpid/acl/AclData.cpp
+++ b/cpp/src/qpid/acl/AclData.cpp
@@ -305,7 +305,9 @@ namespace acl {
     // lookup
     //
     // The ACL main business logic function of matching rules and declaring
-    // an allow or deny result.
+    // an allow or deny result. This lookup is the fastpath per-message
+    // lookup to verify if a user is allowed to publish to an exchange with
+    // a given key.
     //
     AclResult AclData::lookup(
         const std::string&              id,
@@ -331,7 +333,8 @@ namespace acl {
 
             if (itrRule != actionList[action][objType]->end() )
             {
-                //loop the vector
+                // Found a rule list for this user-action-object set.
+                // Search the rule list for a matching rule.
                 ruleSetItr rsItr = itrRule->second.end();
                 for (int cnt = itrRule->second.size(); cnt != 0; cnt--)
                 {
@@ -339,56 +342,46 @@ namespace acl {
 
                     QPID_LOG(debug, "ACL: checking rule " <<  rsItr->toString());
 
-                    // loop the names looking for match
+                    // Search on exchange name and routing key only if specfied in rule.
                     bool match =true;
-                    for (specPropertyMapItr pMItr  = rsItr->props.begin();
-                                           (pMItr != rsItr->props.end()) && match;
-                                            pMItr++)
+                    if (rsItr->pubExchNameInRule)
                     {
-                        //match name is exists first
-                        switch (pMItr->first)
+                        if (matchProp(rsItr->pubExchName, name))
                         {
-                        case acl::SPECPROP_NAME:
-                            if (matchProp(pMItr->second, name))
-                            {
-                                QPID_LOG(debug, "ACL: lookup exchange name '"
-                                    << name << "' matched with rule name '"
-                                    << pMItr->second << "'");
-
-                            }
-                            else
-                            {
-                                match= false;
-                                QPID_LOG(debug, "ACL: lookup exchange name '"
-                                    << name << "' did not match with rule name '"
-                                    << pMItr->second << "'");
-                            }
-                            break;
+                            QPID_LOG(debug, "ACL: Rule: " << rsItr->rawRuleNum << " lookup exchange name '"
+                                << name << "' matched with rule name '"
+                                << rsItr->pubExchName << "'");
 
-                        case acl::SPECPROP_ROUTINGKEY:
-                            if (matchProp(pMItr->second, routingKey))
-                            {
-                                QPID_LOG(debug, "ACL: lookup key name '"
-                                    << routingKey << "' matched with rule routing key '"
-                                    << pMItr->second << "'");
-                            }
-                            else
-                            {
-                                match= false;
-                                QPID_LOG(debug, "ACL: lookup key name '"
-                                    << routingKey << "' did not match with rule routing key '"
-                                    << pMItr->second << "'");
-                            }
-                            break;
+                        }
+                        else
+                        {
+                            match= false;
+                            QPID_LOG(debug, "ACL: Rule: " << rsItr->rawRuleNum << " lookup exchange name '"
+                                << name << "' did not match with rule name '"
+                                << rsItr->pubExchName << "'");
+                        }
+                    }
 
-                        default:
-                            // Don't care
-                            break;
-                        };
+                    if (match && rsItr->pubRoutingKeyInRule)
+                    {
+                        if (rsItr->matchRoutingKey(routingKey))
+                        {
+                            QPID_LOG(debug, "ACL: Rule: " << rsItr->rawRuleNum << " lookup key name '"
+                                << routingKey << "' matched with rule routing key '"
+                                << rsItr->pubRoutingKey << "'");
+                        }
+                        else
+                        {
+                            QPID_LOG(debug, "ACL: Rule: " << rsItr->rawRuleNum << " lookup key name '"
+                                << routingKey << "' did not match with rule routing key '"
+                                << rsItr->pubRoutingKey << "'");
+                            match = false;
+                        }
                     }
+
                     if (match){
                         aclresult = rsItr->ruleMode;
-                        QPID_LOG(debug,"ACL: Successful match, the decision is:"
+                        QPID_LOG(debug,"ACL: Rule: " << rsItr->rawRuleNum << " Successful match, the decision is:"
                             << AclHelper::getAclResultStr(aclresult));
                         return aclresult;
                     }
diff --git a/cpp/src/qpid/acl/AclData.h b/cpp/src/qpid/acl/AclData.h
index 1c1cb3e9c6..ca0a676a1c 100644
--- a/cpp/src/qpid/acl/AclData.h
+++ b/cpp/src/qpid/acl/AclData.h
@@ -21,6 +21,9 @@
  */
 
 #include "qpid/broker/AclModule.h"
+#include "AclTopicMatch.h"
+#include "qpid/log/Statement.h"
+#include "boost/shared_ptr.hpp"
 #include <vector>
 #include <sstream>
 
@@ -48,18 +51,29 @@ public:
     // A single ACL file entry may create many rule entries in
     //  many ruleset vectors.
     //
-    struct rule {
+    struct Rule {
+        typedef broker::TopicExchange::TopicExchangeTester topicTester;
 
         int                   rawRuleNum;   // rule number in ACL file
         qpid::acl::AclResult  ruleMode;     // combined allow/deny log/nolog
         specPropertyMap       props;        //
+        bool                  pubRoutingKeyInRule;
+        std::string           pubRoutingKey;
+        boost::shared_ptr<topicTester> pTTest;
+        bool                  pubExchNameInRule;
+        std::string           pubExchName;
 
-
-        rule (int ruleNum, qpid::acl::AclResult res, specPropertyMap& p) :
+        Rule (int ruleNum, qpid::acl::AclResult res, specPropertyMap& p) :
             rawRuleNum(ruleNum),
             ruleMode(res),
-            props(p)
-            {};
+            props(p),
+            pubRoutingKeyInRule(false),
+            pubRoutingKey(),
+            pTTest(boost::shared_ptr<topicTester>(new topicTester())),
+            pubExchNameInRule(false),
+            pubExchName()
+            {}
+
 
         std::string toString () const {
             std::ostringstream ruleStr;
@@ -76,9 +90,21 @@ public:
             ruleStr << " }]";
             return ruleStr.str();
         }
+
+        void addTopicTest(const std::string& pattern) {
+            pTTest->addBindingKey(broker::TopicExchange::normalize(pattern));
+        }
+
+        // Topic Exchange tester
+        // return true if any bindings match 'pattern'
+        bool matchRoutingKey(const std::string& pattern) const
+        {
+            topicTester::BindingVec bv;
+            return pTTest->findMatches(pattern, bv);
+        }
     };
 
-    typedef  std::vector<rule>               ruleSet;
+    typedef  std::vector<Rule>               ruleSet;
     typedef  ruleSet::const_iterator         ruleSetItr;
     typedef  std::map<std::string, ruleSet > actionObject; // user
     typedef  actionObject::iterator          actObjItr;
diff --git a/cpp/src/qpid/acl/AclReader.cpp b/cpp/src/qpid/acl/AclReader.cpp
index 0830f3f4f4..f9be49b88d 100644
--- a/cpp/src/qpid/acl/AclReader.cpp
+++ b/cpp/src/qpid/acl/AclReader.cpp
@@ -101,7 +101,7 @@ namespace acl {
                         << AclHelper::getAclResultStr(d->decisionMode));
                     foundmode = true;
             } else {
-                AclData::rule rule(cnt, (*i)->res, (*i)->props);
+                AclData::Rule rule(cnt, (*i)->res, (*i)->props);
 
                 // Action -> Object -> map<user -> set<Rule> >
                 std::ostringstream actionstr;
@@ -110,8 +110,27 @@ namespace acl {
                     (*i)->actionAll ? acnt++ : acnt = acl::ACTIONSIZE) {
 
                     if (acnt == acl::ACT_PUBLISH)
+                    {
                         d->transferAcl = true; // we have transfer ACL
-
+                        // For Publish the only object should be Exchange
+                        // and the only property should be routingkey.
+                        // Go through the rule properties and find the name and the key.
+                        // If found then place them specially for the lookup engine.
+                        for (pmCitr pItr=(*i)->props.begin(); pItr!=(*i)->props.end(); pItr++) {
+                            if (acl::SPECPROP_ROUTINGKEY == pItr->first)
+                            {
+                                rule.pubRoutingKeyInRule = true;
+                                rule.pubRoutingKey = (std::string)pItr->second;
+                                rule.addTopicTest(rule.pubRoutingKey);
+                                break;
+                            }
+                            if (acl::SPECPROP_NAME == pItr->first)
+                            {
+                                rule.pubExchNameInRule = true;
+                                rule.pubExchName = pItr->second;
+                            }
+                        }
+                    }
                     actionstr << AclHelper::getActionStr((Action) acnt) << ",";
 
                     //find the Action, create if not exist
@@ -285,7 +304,7 @@ namespace acl {
             if (ws) {
                 ret = true;
             } else {
-                errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber 
+                errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber
                     << ", Non-continuation line must start with \"group\" or \"acl\".";
                 ret = false;
             }
@@ -330,7 +349,7 @@ namespace acl {
         } else {
             const unsigned minimumSize = (cont ? 2 : 3);
             if (toksSize < minimumSize) {
-                errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber 
+                errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber
                     << ", Insufficient tokens for group definition.";
                 return false;
             }
@@ -479,7 +498,7 @@ namespace acl {
                 nvPair propNvp = splitNameValuePair(toks[i]);
                 if (propNvp.second.size() == 0) {
                     errorStream << ACL_FORMAT_ERR_LOG_PREFIX <<  "Line : " << lineNumber
-                        <<", Badly formed property name-value pair \"" 
+                        <<", Badly formed property name-value pair \""
                         << propNvp.first << "\". (Must be name=value)";
                     return false;
                 }
diff --git a/cpp/src/qpid/acl/AclReader.h b/cpp/src/qpid/acl/AclReader.h
index 730013f4ed..6351c1e509 100644
--- a/cpp/src/qpid/acl/AclReader.h
+++ b/cpp/src/qpid/acl/AclReader.h
@@ -26,6 +26,7 @@
 #include <string>
 #include <vector>
 #include <sstream>
+#include <memory>
 #include "qpid/acl/AclData.h"
 #include "qpid/broker/AclModule.h"
 
diff --git a/cpp/src/qpid/acl/AclValidator.cpp b/cpp/src/qpid/acl/AclValidator.cpp
index 49bb65db4b..85f0f7c240 100644
--- a/cpp/src/qpid/acl/AclValidator.cpp
+++ b/cpp/src/qpid/acl/AclValidator.cpp
@@ -131,7 +131,7 @@ namespace acl {
             boost::bind(&AclValidator::validateRule, this, _1));
     }
 
-    void AclValidator::validateRule(qpid::acl::AclData::rule& rule){
+    void AclValidator::validateRule(qpid::acl::AclData::Rule& rule){
         std::for_each(rule.props.begin(),
             rule.props.end(),
             boost::bind(&AclValidator::validateProperty, this, _1));
diff --git a/cpp/src/qpid/acl/AclValidator.h b/cpp/src/qpid/acl/AclValidator.h
index f85c241b06..76eb222d2f 100644
--- a/cpp/src/qpid/acl/AclValidator.h
+++ b/cpp/src/qpid/acl/AclValidator.h
@@ -71,7 +71,7 @@ class AclValidator {
 public:
 
    void validateRuleSet(std::pair<const std::string, qpid::acl::AclData::ruleSet>& rules);
-   void validateRule(qpid::acl::AclData::rule& rule);
+   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);
    AclValidator();