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

7 files changed, 215 insertions, 58 deletions

diff --git a/qpid/cpp/src/qpid/acl/AclData.cpp b/qpid/cpp/src/qpid/acl/AclData.cpp
index da7f240a9b..a07176dc76 100644
--- a/qpid/cpp/src/qpid/acl/AclData.cpp
+++ b/qpid/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/qpid/cpp/src/qpid/acl/AclData.h b/qpid/cpp/src/qpid/acl/AclData.h
index 1c1cb3e9c6..ca0a676a1c 100644
--- a/qpid/cpp/src/qpid/acl/AclData.h
+++ b/qpid/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/qpid/cpp/src/qpid/acl/AclReader.cpp b/qpid/cpp/src/qpid/acl/AclReader.cpp
index 0830f3f4f4..f9be49b88d 100644
--- a/qpid/cpp/src/qpid/acl/AclReader.cpp
+++ b/qpid/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/qpid/cpp/src/qpid/acl/AclReader.h b/qpid/cpp/src/qpid/acl/AclReader.h
index 730013f4ed..6351c1e509 100644
--- a/qpid/cpp/src/qpid/acl/AclReader.h
+++ b/qpid/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/qpid/cpp/src/qpid/acl/AclValidator.cpp b/qpid/cpp/src/qpid/acl/AclValidator.cpp
index 49bb65db4b..85f0f7c240 100644
--- a/qpid/cpp/src/qpid/acl/AclValidator.cpp
+++ b/qpid/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/qpid/cpp/src/qpid/acl/AclValidator.h b/qpid/cpp/src/qpid/acl/AclValidator.h
index f85c241b06..76eb222d2f 100644
--- a/qpid/cpp/src/qpid/acl/AclValidator.h
+++ b/qpid/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();
diff --git a/qpid/cpp/src/tests/acl.py b/qpid/cpp/src/tests/acl.py
index 663af3e8ac..0e096a6f5b 100755
--- a/qpid/cpp/src/tests/acl.py
+++ b/qpid/cpp/src/tests/acl.py
@@ -310,7 +310,7 @@ class ACLTests(TestBase010):
                 self.fail("ACL should allow queue create request");
             self.fail("Error during queue create request");
 
-        
+
 
     def test_user_realm(self):
         """
@@ -1537,6 +1537,124 @@ class ACLTests(TestBase010):
 
 
    #=====================================
+   # QMF Topic Exchange tests
+   #=====================================
+
+    def test_qmf_topic_exchange_tests(self):
+        """
+        Test using QMF method hooks into ACL logic
+        """
+        aclf = self.get_acl_file()
+        aclf.write('# begin hack alert: allow anonymous to access the lookup debug functions\n')
+        aclf.write('acl allow-log anonymous create  queue\n')
+        aclf.write('acl allow-log anonymous all     exchange name=qmf.*\n')
+        aclf.write('acl allow-log anonymous all     exchange name=amq.direct\n')
+        aclf.write('acl allow-log anonymous all     exchange name=qpid.management\n')
+        aclf.write('acl allow-log anonymous access  method   name=*\n')
+        aclf.write('# end hack alert\n')
+        aclf.write('acl allow-log uPlain1@COMPANY   publish exchange name=X routingkey=ab.cd.e\n')
+        aclf.write('acl allow-log uPlain2@COMPANY   publish exchange name=X routingkey=.\n')
+        aclf.write('acl allow-log uStar1@COMPANY    publish exchange name=X routingkey=a.*.b\n')
+        aclf.write('acl allow-log uStar2@COMPANY    publish exchange name=X routingkey=*.x\n')
+        aclf.write('acl allow-log uStar3@COMPANY    publish exchange name=X routingkey=x.x.*\n')
+        aclf.write('acl allow-log uHash1@COMPANY    publish exchange name=X routingkey=a.#.b\n')
+        aclf.write('acl allow-log uHash2@COMPANY    publish exchange name=X routingkey=a.#\n')
+        aclf.write('acl allow-log uHash3@COMPANY    publish exchange name=X routingkey=#.a\n')
+        aclf.write('acl allow-log uHash4@COMPANY    publish exchange name=X routingkey=a.#.b.#.c\n')
+        aclf.write('acl allow-log uMixed1@COMPANY   publish exchange name=X routingkey=*.x.#.y\n')
+        aclf.write('acl allow-log uMixed2@COMPANY   publish exchange name=X routingkey=a.#.b.*\n')
+        aclf.write('acl allow-log uMixed3@COMPANY   publish exchange name=X routingkey=*.*.*.#\n')
+
+        aclf.write('acl allow-log all publish exchange name=X routingkey=MN.OP.Q\n')
+        aclf.write('acl allow-log all publish exchange name=X routingkey=M.*.N\n')
+        aclf.write('acl allow-log all publish exchange name=X routingkey=M.#.N\n')
+        aclf.write('acl allow-log all publish exchange name=X routingkey=*.M.#.N\n')
+
+        aclf.write('acl deny-log all all\n')
+        aclf.close()
+
+        result = self.reload_acl()
+        if (result):
+            self.fail(result)
+
+        #                                  aclKey: "ab.cd.e"
+        self.LookupPublish("uPlain1@COMPANY", "X", "ab.cd.e",   "allow-log")
+        self.LookupPublish("uPlain1@COMPANY", "X", "abx.cd.e",  "deny-log")
+        self.LookupPublish("uPlain1@COMPANY", "X", "ab.cd",     "deny-log")
+        self.LookupPublish("uPlain1@COMPANY", "X", "ab.cd..e.", "deny-log")
+        self.LookupPublish("uPlain1@COMPANY", "X", "ab.cd.e.",  "deny-log")
+        self.LookupPublish("uPlain1@COMPANY", "X", ".ab.cd.e",  "deny-log")
+        #                                  aclKey: "."
+        self.LookupPublish("uPlain2@COMPANY", "X", ".",         "allow-log")
+
+        #                                  aclKey: "a.*.b"
+        self.LookupPublish("uStar1@COMPANY", "X", "a.xx.b",   "allow-log")
+        self.LookupPublish("uStar1@COMPANY", "X", "a.b",      "deny-log")
+        #                                  aclKey: "*.x"
+        self.LookupPublish("uStar2@COMPANY", "X", "y.x",      "allow-log")
+        self.LookupPublish("uStar2@COMPANY", "X", ".x",       "allow-log")
+        self.LookupPublish("uStar2@COMPANY", "X", "x",        "deny-log")
+        #                                  aclKey: "x.x.*"
+        self.LookupPublish("uStar3@COMPANY", "X", "x.x.y",      "allow-log")
+        self.LookupPublish("uStar3@COMPANY", "X", "x.x.",       "allow-log")
+        self.LookupPublish("uStar3@COMPANY", "X", "x.x",        "deny-log")
+        self.LookupPublish("uStar3@COMPANY", "X", "q.x.y",      "deny-log")
+
+        #                                  aclKey: "a.#.b"
+        self.LookupPublish("uHash1@COMPANY", "X", "a.b",         "allow-log")
+        self.LookupPublish("uHash1@COMPANY", "X", "a.x.b",       "allow-log")
+        self.LookupPublish("uHash1@COMPANY", "X", "a..x.y.zz.b", "allow-log")
+        self.LookupPublish("uHash1@COMPANY", "X", "a.b.",        "deny-log")
+        self.LookupPublish("uHash1@COMPANY", "X", "q.x.b",       "deny-log")
+
+        #                                  aclKey: "a.#"
+        self.LookupPublish("uHash2@COMPANY", "X", "a",         "allow-log")
+        self.LookupPublish("uHash2@COMPANY", "X", "a.b",       "allow-log")
+        self.LookupPublish("uHash2@COMPANY", "X", "a.b.c",     "allow-log")
+
+        #                                  aclKey: "#.a"
+        self.LookupPublish("uHash3@COMPANY", "X", "a",         "allow-log")
+        self.LookupPublish("uHash3@COMPANY", "X", "x.y.a",     "allow-log")
+
+        #                                  aclKey: "a.#.b.#.c"
+        self.LookupPublish("uHash4@COMPANY", "X", "a.b.c",         "allow-log")
+        self.LookupPublish("uHash4@COMPANY", "X", "a.x.b.y.c",     "allow-log")
+        self.LookupPublish("uHash4@COMPANY", "X", "a.x.x.b.y.y.c", "allow-log")
+
+        #                                  aclKey: "*.x.#.y"
+        self.LookupPublish("uMixed1@COMPANY", "X", "a.x.y",          "allow-log")
+        self.LookupPublish("uMixed1@COMPANY", "X", "a.x.p.qq.y",     "allow-log")
+        self.LookupPublish("uMixed1@COMPANY", "X", "a.a.x.y",        "deny-log")
+        self.LookupPublish("uMixed1@COMPANY", "X", "aa.x.b.c",       "deny-log")
+
+        #                                  aclKey: "a.#.b.*"
+        self.LookupPublish("uMixed2@COMPANY", "X", "a.b.x",          "allow-log")
+        self.LookupPublish("uMixed2@COMPANY", "X", "a.x.x.x.b.x",    "allow-log")
+
+        #                                  aclKey: "*.*.*.#"
+        self.LookupPublish("uMixed3@COMPANY", "X", "x.y.z",          "allow-log")
+        self.LookupPublish("uMixed3@COMPANY", "X", "x.y.z.a.b.c",    "allow-log")
+        self.LookupPublish("uMixed3@COMPANY", "X", "x.y",            "deny-log")
+        self.LookupPublish("uMixed3@COMPANY", "X", "x",              "deny-log")
+
+        # Repeat the keys with wildcard user spec
+        self.LookupPublish("uPlain1@COMPANY", "X", "MN.OP.Q",        "allow-log")
+        self.LookupPublish("uStar1@COMPANY" , "X", "M.xx.N",         "allow-log")
+        self.LookupPublish("uHash1@COMPANY" , "X", "M.N",            "allow-log")
+        self.LookupPublish("uHash1@COMPANY" , "X", "M.x.N",          "allow-log")
+        self.LookupPublish("uHash1@COMPANY" , "X", "M..x.y.zz.N",    "allow-log")
+        self.LookupPublish("uMixed1@COMPANY", "X", "a.M.N",          "allow-log")
+        self.LookupPublish("uMixed1@COMPANY", "X", "a.M.p.qq.N",     "allow-log")
+
+        self.LookupPublish("dev@QPID", "X", "MN.OP.Q",        "allow-log")
+        self.LookupPublish("dev@QPID", "X", "M.xx.N",         "allow-log")
+        self.LookupPublish("dev@QPID", "X", "M.N",            "allow-log")
+        self.LookupPublish("dev@QPID", "X", "M.x.N",          "allow-log")
+        self.LookupPublish("dev@QPID", "X", "M..x.y.zz.N",    "allow-log")
+        self.LookupPublish("dev@QPID", "X", "a.M.N",          "allow-log")
+        self.LookupPublish("dev@QPID", "X", "a.M.p.qq.N",     "allow-log")
+
+   #=====================================
    # Connection limits
    #=====================================