I have applied the patch attached to QPID-2108 from Tim Platten with a few modifications.

(Please refer to the JIRA for a description of these modifications)
I also fixed the formatting and added more debug loggging for the publish acl lookup method in AclData.cpp
I also added test cases to cover QPID-2108


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

4 files changed, 214 insertions, 49 deletions

diff --git a/qpid/cpp/src/qpid/acl/AclData.cpp b/qpid/cpp/src/qpid/acl/AclData.cpp
index 81519c3311..1b3cdea028 100644
--- a/qpid/cpp/src/qpid/acl/AclData.cpp
+++ b/qpid/cpp/src/qpid/acl/AclData.cpp
@@ -18,7 +18,7 @@
 
 #include "qpid/acl/AclData.h"
 #include "qpid/log/Statement.h"
-
+#include <boost/lexical_cast.hpp>
 
 namespace qpid {
 namespace acl {
@@ -57,14 +57,15 @@ AclResult AclData::lookup(const std::string& id, const Action& action, const Obj
 		const std::string& name, std::map<Property, std::string>* params) {
 
 	QPID_LOG(debug, "ACL: Lookup for id:" << id << " action:" << AclHelper::getActionStr((Action) action)
-			<< " objectType:" << AclHelper::getObjectTypeStr((ObjectType) objType) << " name:" << name
-			<< " with params " << AclHelper::propertyMapToString(params));
+		 << " objectType:" << AclHelper::getObjectTypeStr((ObjectType) objType) << " name:" << name
+		 << " with params " << AclHelper::propertyMapToString(params));
 
 	AclResult aclresult = decisionMode;
 	if (actionList[action] && actionList[action][objType]) {
 		AclData::actObjItr itrRule = actionList[action][objType]->find(id);
 		if (itrRule == actionList[action][objType]->end())
 			itrRule = actionList[action][objType]->find("*");
+
 		if (itrRule != actionList[action][objType]->end()) {
 
 			QPID_LOG(debug, "ACL: checking the following rules for : " << itrRule->first );
@@ -79,25 +80,48 @@ AclResult AclData::lookup(const std::string& id, const Action& action, const Obj
 					if (pMItr->first == acl::PROP_NAME) {
 						if (matchProp(pMItr->second, name)){
 							QPID_LOG(debug, "ACL: name '" << name << "' matched with name '"
-																        << pMItr->second << "' given in the rule");
-					    }else{
+								 << pMItr->second << "' given in the rule");
+					        }else{
 							match = false;
 							QPID_LOG(debug, "ACL: name '" << name << "' didn't match with name '"
-									        << pMItr->second << "' given in the rule");
+								 << pMItr->second << "' given in the rule");
 						}
 					} else if (params) { //match pMItr against params
 						propertyMapItr paramItr = params->find(pMItr->first);
 						if (paramItr == params->end()) {
 							match = false;
 							QPID_LOG(debug, "ACL: the given parameter map in lookup doesn't contain the property '"
-									        << AclHelper::getPropertyStr(pMItr->first) << "'");
-						} else if (!matchProp(pMItr->second, paramItr->second)) {
+								 << AclHelper::getPropertyStr(pMItr->first) << "'");
+						}else if ( pMItr->first == acl::PROP_MAXQUEUECOUNT || pMItr->first == acl::PROP_MAXQUEUESIZE ) {
+                                                      if ( pMItr->first == paramItr->first ) {
+                                                          uint64_t aclMax = boost::lexical_cast<uint64_t>(pMItr->second);
+                                                          uint64_t paramMax = boost::lexical_cast<uint64_t>(paramItr->second);
+                                                          QPID_LOG(debug, "ACL: Numeric comparison for property " <<
+                                                                   AclHelper::getPropertyStr(paramItr->first)  <<
+                                                                   " (value given in lookup = " << 
+                                                                   boost::lexical_cast<std::string>(paramItr->second) << 
+                                                                   ", value give in rule = " <<
+                                                                   boost::lexical_cast<std::string>(pMItr->second) << " )");  
+                                                          if (( aclMax ) && ( paramMax == 0 || paramMax > aclMax)){
+                                                              match = decisionMode == qpid::acl::ALLOW ;
+                                                              QPID_LOG(debug, "ACL: Limit exceeded and match=" << 
+                                                              (match ? "true": "false") <<
+                                                              " as decision mode is " << AclHelper::getAclResultStr(decisionMode));
+                                                          }
+                                                      }
+                                                }else if (matchProp(pMItr->second, paramItr->second)) {
+                                                       QPID_LOG(debug, "ACL: the pair("
+								 << AclHelper::getPropertyStr(paramItr->first) << "," << paramItr->second
+								 << ") given in lookup matched the pair("
+								 << AclHelper::getPropertyStr(pMItr->first) << "," << pMItr->second << ") given in the rule"); 
+						} else {
 							QPID_LOG(debug, "ACL: the pair("
-									        << AclHelper::getPropertyStr(paramItr->first) << "," << paramItr->second
-									        << ") given in lookup doesn't match the pair("
-									        << AclHelper::getPropertyStr(pMItr->first) << "," << pMItr->second << ") given in the rule");
+								 << AclHelper::getPropertyStr(paramItr->first) << "," << paramItr->second
+								 << ") given in lookup doesn't match the pair("
+								 << AclHelper::getPropertyStr(pMItr->first) << "," << pMItr->second << ") given in the rule");
 							match = false;
-						}
+                  
+                                                } 
 					}
 				}
 				if (match)
@@ -116,37 +140,63 @@ AclResult AclData::lookup(const std::string& id, const Action& action, const Obj
 
 AclResult AclData::lookup(const std::string& id, const Action& action, const ObjectType& objType, const std::string& /*Exchange*/ name, const std::string& RoutingKey)
 {
-     AclResult aclresult = decisionMode;
+
+	QPID_LOG(debug, "ACL: Lookup for id:" << id << " action:" << AclHelper::getActionStr((Action) action)
+		 << " objectType:" << AclHelper::getObjectTypeStr((ObjectType) objType) << " exchange name:" << name
+		 << " with routing key " << RoutingKey);
+
+        AclResult aclresult = decisionMode;
 	
-	 if (actionList[action] && actionList[action][objType]){
-	      AclData::actObjItr itrRule = actionList[action][objType]->find(id);
-		  if (itrRule == actionList[action][objType]->end())
+	if (actionList[action] && actionList[action][objType]){
+	          AclData::actObjItr itrRule = actionList[action][objType]->find(id);
+		  
+                  if (itrRule == actionList[action][objType]->end())
 		       itrRule = actionList[action][objType]->find("*");
+
 		  if (itrRule != actionList[action][objType]->end() ) {
 			   
+                        QPID_LOG(debug, "ACL: checking the following rules for : " << itrRule->first );
+    
 			   //loop the vector
-    		   for (ruleSetItr i=itrRule->second.begin(); i<itrRule->second.end(); i++) {
-                    
+    		        for (ruleSetItr i=itrRule->second.begin(); i<itrRule->second.end(); i++) {
+				QPID_LOG(debug, "ACL: checking rule " <<  i->toString());
+	                   
 					// loop the names looking for match
 					bool match =true;
 					for (propertyMapItr pMItr = i->props.begin(); (pMItr != i->props.end()) && match; pMItr++)
 					{
-                        //match name is exists first
+                                                //match name is exists first
 						if (pMItr->first == acl::PROP_NAME){
-						     if (!matchProp(pMItr->second, name)){  
-							     match= false;
-							 }
+						     if (matchProp(pMItr->second, name)){  							     
+                                                         QPID_LOG(debug, "ACL: name '" << name << "' matched with name '"
+								  << pMItr->second << "' given in the rule");
+                                                             
+					             }else{
+                                                         match= false;
+                                                         QPID_LOG(debug, "ACL: name '" << name << "' didn't match with name '"
+							          << pMItr->second << "' given in the rule");
+                                                     }    
 						}else if (pMItr->first == acl::PROP_ROUTINGKEY){
-						     if (!matchProp(pMItr->second, RoutingKey)){  
-							     match= false;
-							 }
+						     if (matchProp(pMItr->second, RoutingKey)){  
+							 QPID_LOG(debug, "ACL: name '" << name << "' matched with routing_key '"
+							          << pMItr->second << "' given in the rule");    
+					             }else{
+                                                         match= false;
+                                                         QPID_LOG(debug, "ACL: name '" << name << "' didn't match with routing_key '"
+							          << pMItr->second << "' given in the rule");
+                                                     } 
 						}
 					}
-					if (match) return getACLResult(i->logOnly, i->log);
-    		   }
+					if (match){
+                                             aclresult = getACLResult(i->logOnly, i->log);
+					     QPID_LOG(debug,"Successful match, the decision is:" << AclHelper::getAclResultStr(aclresult));
+					     return aclresult;
+                                        }
+    		        }
 		  }
-	 }
-     return aclresult;
+	}
+	QPID_LOG(debug,"No successful match, defaulting to the decision mode " << AclHelper::getAclResultStr(aclresult));
+        return aclresult;
 
 }
 
diff --git a/qpid/cpp/src/qpid/broker/AclModule.h b/qpid/cpp/src/qpid/broker/AclModule.h
index 536fa21b2b..2f4f7eaacc 100644
--- a/qpid/cpp/src/qpid/broker/AclModule.h
+++ b/qpid/cpp/src/qpid/broker/AclModule.h
@@ -40,7 +40,8 @@ enum Action {ACT_CONSUME, ACT_PUBLISH, ACT_CREATE, ACT_ACCESS, ACT_BIND,
 enum Property {PROP_NAME, PROP_DURABLE, PROP_OWNER, PROP_ROUTINGKEY,
                PROP_PASSIVE, PROP_AUTODELETE, PROP_EXCLUSIVE, PROP_TYPE,
                PROP_ALTERNATE, PROP_QUEUENAME, PROP_SCHEMAPACKAGE,
-               PROP_SCHEMACLASS};
+               PROP_SCHEMACLASS, PROP_POLICYTYPE, PROP_MAXQUEUESIZE,
+               PROP_MAXQUEUECOUNT};
 enum AclResult {ALLOW, ALLOWLOG, DENY, DENYLOG};	
 
 } // namespace acl
@@ -132,6 +133,9 @@ class AclHelper {
         if (str.compare("queuename") == 0) return PROP_QUEUENAME;
         if (str.compare("schemapackage") == 0) return PROP_SCHEMAPACKAGE;
         if (str.compare("schemaclass") == 0) return PROP_SCHEMACLASS;
+        if (str.compare("policytype") == 0) return PROP_POLICYTYPE;
+        if (str.compare("maxqueuesize") == 0) return PROP_MAXQUEUESIZE;
+        if (str.compare("maxqueuecount") == 0) return PROP_MAXQUEUECOUNT; 
         throw str;
     }
     static inline std::string getPropertyStr(const Property p) {
@@ -148,6 +152,9 @@ class AclHelper {
           case PROP_QUEUENAME: return "queuename";
           case PROP_SCHEMAPACKAGE: return "schemapackage";
           case PROP_SCHEMACLASS: return "schemaclass";
+          case PROP_POLICYTYPE: return "policytype";
+          case PROP_MAXQUEUESIZE: return "maxqueuesize";
+          case PROP_MAXQUEUECOUNT: return "maxqueuecount"; 
           default: assert(false); // should never get here
         }
         return "";
@@ -217,11 +224,14 @@ class AclHelper {
         // == Queues ==
 
         propSetPtr p4(new propSet);
-        p3->insert(PROP_ALTERNATE);
-        p3->insert(PROP_PASSIVE);
-        p3->insert(PROP_DURABLE);
-        p3->insert(PROP_EXCLUSIVE);
-        p3->insert(PROP_AUTODELETE);
+        p4->insert(PROP_ALTERNATE);
+        p4->insert(PROP_PASSIVE);
+        p4->insert(PROP_DURABLE);
+        p4->insert(PROP_EXCLUSIVE);
+        p4->insert(PROP_AUTODELETE);
+        p4->insert(PROP_POLICYTYPE);
+        p4->insert(PROP_MAXQUEUESIZE);
+        p4->insert(PROP_MAXQUEUECOUNT);
 
         actionMapPtr a1(new actionMap);
         a1->insert(actionPair(ACT_ACCESS,  p0));
diff --git a/qpid/cpp/src/qpid/broker/SessionAdapter.cpp b/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
index a1ad5a0a30..4ee3a97357 100644
--- a/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
+++ b/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
@@ -337,6 +337,10 @@ void SessionAdapter::QueueHandlerImpl::declare(const string& name, const string&
         params.insert(make_pair(acl::PROP_DURABLE, std::string(durable ? _TRUE : _FALSE)));
         params.insert(make_pair(acl::PROP_EXCLUSIVE, std::string(exclusive ? _TRUE : _FALSE)));
         params.insert(make_pair(acl::PROP_AUTODELETE, std::string(autoDelete ? _TRUE : _FALSE)));
+        params.insert(make_pair(acl::PROP_POLICYTYPE, arguments.getAsString("qpid.policy_type")));
+        params.insert(make_pair(acl::PROP_MAXQUEUECOUNT, arguments.getAsString("qpid.max_count")));
+        params.insert(make_pair(acl::PROP_MAXQUEUESIZE, arguments.getAsString("qpid.max_size")));
+
         if (!acl->authorise(getConnection().getUserId(),acl::ACT_CREATE,acl::OBJ_QUEUE,name,&params) )
             throw NotAllowedException(QPID_MSG("ACL denied queue create request from " << getConnection().getUserId()));
     }
@@ -472,8 +476,7 @@ SessionAdapter::MessageHandlerImpl::subscribe(const string& queueName,
 
     AclModule* acl = getBroker().getAcl();
     if (acl)
-    {
-        // add flags as needed
+    {        
          if (!acl->authorise(getConnection().getUserId(),acl::ACT_CONSUME,acl::OBJ_QUEUE,queueName,NULL) )
              throw NotAllowedException(QPID_MSG("ACL denied Queue subscribe request from " << getConnection().getUserId()));
     }
diff --git a/qpid/cpp/src/tests/acl.py b/qpid/cpp/src/tests/acl.py
index fc53d2ce8b..e6cfdb56d3 100755
--- a/qpid/cpp/src/tests/acl.py
+++ b/qpid/cpp/src/tests/acl.py
@@ -220,10 +220,11 @@ class ACLTests(TestBase010):
         """
         aclf = ACLFile()
         aclf.write('acl deny bob@QPID create queue name=q1 durable=true passive=true\n')
-        aclf.write('acl deny bob@QPID create queue name=q2 exclusive=true\n')
+        aclf.write('acl deny bob@QPID create queue name=q2 exclusive=true policytype=ring\n')
         aclf.write('acl deny bob@QPID access queue name=q3\n')
         aclf.write('acl deny bob@QPID purge queue name=q3\n')
-        aclf.write('acl deny bob@QPID delete queue name=q4\n')                
+        aclf.write('acl deny bob@QPID delete queue name=q4\n')  
+        aclf.write('acl deny bob@QPID create queue name=q5 maxqueuesize=1000 maxqueuecount=100\n')              
         aclf.write('acl allow all all')
         aclf.close()        
         
@@ -241,19 +242,41 @@ class ACLTests(TestBase010):
             session = self.get_session('bob','bob')
         
         try:
-            session.queue_declare(queue="q2", exclusive=True)
-            self.fail("ACL should deny queue create request with name=q2 exclusive=true");
+            queue_options = {}
+            queue_options["qpid.policy_type"] = "ring" 
+            session.queue_declare(queue="q2", exclusive=True, arguments=queue_options)
+            self.fail("ACL should deny queue create request with name=q2 exclusive=true qpid.policy_type=ring");
         except qpid.session.SessionException, e:
             self.assertEqual(530,e.args[0].error_code) 
             session = self.get_session('bob','bob')
         
         try:
-            session.queue_declare(queue="q2", durable=True)            
+            queue_options = {}
+            queue_options["qpid.policy_type"] = "ring_strict"   
+            session.queue_declare(queue="q2", exclusive=True, arguments=queue_options)            
         except qpid.session.SessionException, e:
             if (530 == e.args[0].error_code):
-                self.fail("ACL should allow queue create request for q2 with any parameter other than exclusive=true");
+                self.fail("ACL should allow queue create request with name=q2 exclusive=true qpid.policy_type=ring_strict");
 
         try:
+            queue_options = {}
+            queue_options["qpid.max_count"] = "200"
+            queue_options["qpid.max_size"] = "500"  
+            session.queue_declare(queue="q5", exclusive=True, arguments=queue_options)
+            self.fail("ACL should deny queue create request with name=q2, qpid.max_size=500 and qpid.max_count=200");
+        except qpid.session.SessionException, e:
+            self.assertEqual(530,e.args[0].error_code) 
+            session = self.get_session('bob','bob')
+
+        try:
+            queue_options = {}
+            queue_options["qpid.max_count"] = "200"
+            queue_options["qpid.max_size"] = "100"  
+            session.queue_declare(queue="q2", exclusive=True, arguments=queue_options)
+        except qpid.session.SessionException, e:
+            if (530 == e.args[0].error_code):
+                self.fail("ACL should allow queue create request with name=q2, qpid.max_size=100 and qpid.max_count=200 ");
+        try:
             session.queue_declare(queue="q3", exclusive=True)
             session.queue_declare(queue="q4", durable=True)
         except qpid.session.SessionException, e:
@@ -300,12 +323,13 @@ class ACLTests(TestBase010):
         """
         aclf = ACLFile()
         aclf.write('acl allow bob@QPID create queue name=q1 durable=true passive=true\n')
-        aclf.write('acl allow bob@QPID create queue name=q2 exclusive=true\n')
+        aclf.write('acl allow bob@QPID create queue name=q2 exclusive=true policytype=ring\n')
         aclf.write('acl allow bob@QPID access queue name=q3\n')
         aclf.write('acl allow bob@QPID purge queue name=q3\n')
         aclf.write('acl allow bob@QPID create queue name=q3\n')                
         aclf.write('acl allow bob@QPID create queue name=q4\n')                
-        aclf.write('acl allow bob@QPID delete queue name=q4\n')                
+        aclf.write('acl allow bob@QPID delete queue name=q4\n')   
+        aclf.write('acl allow bob@QPID create queue name=q5 maxqueuesize=1000 maxqueuecount=100\n')                   
         aclf.write('acl allow guest@QPID all all\n')
         aclf.write('acl deny all all')
         aclf.close()        
@@ -337,10 +361,31 @@ class ACLTests(TestBase010):
             session = self.get_session('bob','bob')
         
         try:
-            session.queue_declare(queue="q2", exclusive=True)            
+            queue_options = {}
+            queue_options["qpid.max_count"] = "200"
+            queue_options["qpid.max_size"] = "500"  
+            session.queue_declare(queue="q5", arguments=queue_options)
+            self.fail("ACL should deny queue create request with name=q2 maxqueuesize=500 maxqueuecount=200");
+        except qpid.session.SessionException, e:
+            self.assertEqual(530,e.args[0].error_code) 
+            session = self.get_session('bob','bob')
+
+        try:
+            queue_options = {}
+            queue_options["qpid.max_count"] = "100"
+            queue_options["qpid.max_size"] = "500"  
+            session.queue_declare(queue="q5", arguments=queue_options)  
         except qpid.session.SessionException, e:
             if (530 == e.args[0].error_code):
-                self.fail("ACL should allow queue create request for q2 with exclusive=true");
+                self.fail("ACL should allow queue create request with name=q2 maxqueuesize=500 maxqueuecount=200");
+
+        try:
+            queue_options = {}
+            queue_options["qpid.policy_type"] = "ring"
+            session.queue_declare(queue="q2", exclusive=True, arguments=queue_options)            
+        except qpid.session.SessionException, e:
+            if (530 == e.args[0].error_code):
+                self.fail("ACL should allow queue create request for q2 with exclusive=true policytype=ring");
 
         try:
             session.queue_declare(queue="q3")
@@ -733,7 +778,7 @@ class ACLTests(TestBase010):
    # ACL publish tests
    #=====================================
    
-    def test_publish_acl(self):
+    def test_publish_acl_allow_mode(self):
         """
         Test various publish acl
         """
@@ -779,4 +824,61 @@ class ACLTests(TestBase010):
             session.message_transfer(destination="amq.direct", message=Message(props,"Test"))
         except qpid.session.SessionException, e:
             if (530 == e.args[0].error_code):
-                self.fail("ACL should allow message transfer to exchange amq.direct");
+                self.fail("ACL should allow message transfer to exchange amq.direct with routing key rk2");
+
+
+    def test_publish_acl_deny_mode(self):
+        """
+        Test various publish acl
+        """
+        aclf = ACLFile()
+        aclf.write('acl allow bob@QPID publish exchange name=amq.direct routingkey=rk1\n')
+        aclf.write('acl allow bob@QPID publish exchange name=amq.topic\n')
+        aclf.write('acl allow bob@QPID publish exchange name=myEx routingkey=rk2\n')
+        aclf.write('acl allow bob@QPID create exchange\n')    
+        aclf.write('acl allow guest@QPID all all \n')    
+        aclf.write('acl deny all all')
+        aclf.close()        
+        
+        result = self.reload_acl()
+        if (result.text.find("format error",0,len(result.text)) != -1):
+            self.fail(result)        
+        
+        session = self.get_session('bob','bob')
+            
+        props = session.delivery_properties(routing_key="rk2")
+               
+        try:            
+            session.message_transfer(destination="amq.direct", message=Message(props,"Test"))
+            self.fail("ACL should deny message transfer to name=amq.direct routingkey=rk2");
+        except qpid.session.SessionException, e:
+            self.assertEqual(530,e.args[0].error_code)
+            session = self.get_session('bob','bob')                        
+            
+        try:
+            session.message_transfer(destination="amq.topic", message=Message(props,"Test"))
+        except qpid.session.SessionException, e:
+            if (530 == e.args[0].error_code):
+                self.fail("ACL should allow message transfer to exchange amq.topic with any routing key");               
+                        
+        try:
+            session.exchange_declare(exchange='myEx', type='direct', durable=False)
+            session.message_transfer(destination="myEx", message=Message(props,"Test"))
+        except qpid.session.SessionException, e:
+            if (530 == e.args[0].error_code):
+                self.fail("ACL should allow message transfer to exchange myEx with routing key=rk2");               
+                        
+        props = session.delivery_properties(routing_key="rk1")
+
+        try:
+            session.message_transfer(destination="myEx", message=Message(props,"Test"))
+            self.fail("ACL should deny message transfer to name=myEx routingkey=rk1");
+        except qpid.session.SessionException, e:
+            self.assertEqual(530,e.args[0].error_code)
+            session = self.get_session('bob','bob')                                                
+
+        try:
+            session.message_transfer(destination="amq.direct", message=Message(props,"Test"))
+        except qpid.session.SessionException, e:
+            if (530 == e.args[0].error_code):
+                self.fail("ACL should allow message transfer to exchange amq.direct with routing key rk1");