QPID-4268 C++ Broker Acl support for limiting on-disk store file size and count

Reviewed at https://reviews.apache.org/r/6838/



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

7 files changed, 345 insertions, 1 deletions

diff --git a/cpp/src/qpid/acl/AclData.cpp b/cpp/src/qpid/acl/AclData.cpp
index ac97e53b14..997fbdf2eb 100644
--- a/cpp/src/qpid/acl/AclData.cpp
+++ b/cpp/src/qpid/acl/AclData.cpp
@@ -199,6 +199,16 @@ namespace acl {
                                     lookupParamItr = params->find(PROP_MAXQUEUESIZE);
                                     break;
 
+                                case acl::SPECPROP_MAXFILECOUNTUPPERLIMIT:
+                                case acl::SPECPROP_MAXFILECOUNTLOWERLIMIT:
+                                    lookupParamItr = params->find(PROP_MAXFILECOUNT);
+                                    break;
+
+                                case acl::SPECPROP_MAXFILESIZEUPPERLIMIT:
+                                case acl::SPECPROP_MAXFILESIZELOWERLIMIT:
+                                    lookupParamItr = params->find(PROP_MAXFILESIZE);
+                                    break;
+
                                 default:
                                     lookupParamItr = params->find((Property)rulePropMapItr->first);
                                     break;
@@ -222,6 +232,8 @@ namespace acl {
                                     {
                                     case acl::SPECPROP_MAXQUEUECOUNTUPPERLIMIT:
                                     case acl::SPECPROP_MAXQUEUESIZEUPPERLIMIT:
+                                    case acl::SPECPROP_MAXFILECOUNTUPPERLIMIT:
+                                    case acl::SPECPROP_MAXFILESIZEUPPERLIMIT:
                                         limitChecked &=
                                             compareIntMax(
                                                 rulePropMapItr->first,
@@ -231,6 +243,8 @@ namespace acl {
 
                                     case acl::SPECPROP_MAXQUEUECOUNTLOWERLIMIT:
                                     case acl::SPECPROP_MAXQUEUESIZELOWERLIMIT:
+                                    case acl::SPECPROP_MAXFILECOUNTLOWERLIMIT:
+                                    case acl::SPECPROP_MAXFILESIZELOWERLIMIT:
                                         limitChecked &=
                                             compareIntMin(
                                                 rulePropMapItr->first,
diff --git a/cpp/src/qpid/acl/AclValidator.cpp b/cpp/src/qpid/acl/AclValidator.cpp
index 85f0f7c240..73b49b2959 100644
--- a/cpp/src/qpid/acl/AclValidator.cpp
+++ b/cpp/src/qpid/acl/AclValidator.cpp
@@ -94,6 +94,22 @@ namespace acl {
                                     boost::shared_ptr<PropertyType>(
                                         new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
 
+        validators.insert(Validator(acl::SPECPROP_MAXFILESIZELOWERLIMIT,
+                                    boost::shared_ptr<PropertyType>(
+                                        new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
+        validators.insert(Validator(acl::SPECPROP_MAXFILESIZEUPPERLIMIT,
+                                    boost::shared_ptr<PropertyType>(
+                                        new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
+        validators.insert(Validator(acl::SPECPROP_MAXFILECOUNTLOWERLIMIT,
+                                    boost::shared_ptr<PropertyType>(
+                                        new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
+        validators.insert(Validator(acl::SPECPROP_MAXFILECOUNTUPPERLIMIT,
+                                    boost::shared_ptr<PropertyType>(
+                                        new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
         std::string policyTypes[] = {"ring", "ring_strict", "flow_to_disk", "reject"};
         std::vector<std::string> v(policyTypes, policyTypes + sizeof(policyTypes) / sizeof(std::string));
         validators.insert(Validator(acl::SPECPROP_POLICYTYPE,
diff --git a/cpp/src/qpid/broker/AclModule.h b/cpp/src/qpid/broker/AclModule.h
index 987d3e4a78..4070e49559 100644
--- a/cpp/src/qpid/broker/AclModule.h
+++ b/cpp/src/qpid/broker/AclModule.h
@@ -79,6 +79,8 @@ namespace acl {
         PROP_POLICYTYPE,
         PROP_MAXQUEUESIZE,
         PROP_MAXQUEUECOUNT,
+        PROP_MAXFILESIZE,
+        PROP_MAXFILECOUNT,
         PROPERTYSIZE           // PROPERTYSIZE must be last in list
     };
 
@@ -102,7 +104,11 @@ namespace acl {
         SPECPROP_MAXQUEUESIZELOWERLIMIT,
         SPECPROP_MAXQUEUESIZEUPPERLIMIT,
         SPECPROP_MAXQUEUECOUNTLOWERLIMIT,
-        SPECPROP_MAXQUEUECOUNTUPPERLIMIT };
+        SPECPROP_MAXQUEUECOUNTUPPERLIMIT,
+        SPECPROP_MAXFILESIZELOWERLIMIT,
+        SPECPROP_MAXFILESIZEUPPERLIMIT,
+        SPECPROP_MAXFILECOUNTLOWERLIMIT,
+        SPECPROP_MAXFILECOUNTUPPERLIMIT };
 
 // AclResult  shared between ACL spec and ACL authorise interface
     enum AclResult {
@@ -227,6 +233,8 @@ namespace acl {
             if (str.compare("policytype")    == 0) return PROP_POLICYTYPE;
             if (str.compare("maxqueuesize")  == 0) return PROP_MAXQUEUESIZE;
             if (str.compare("maxqueuecount") == 0) return PROP_MAXQUEUECOUNT;
+            if (str.compare("maxfilesize")   == 0) return PROP_MAXFILESIZE;
+            if (str.compare("maxfilecount")  == 0) return PROP_MAXFILECOUNT;
             throw qpid::Exception(str);
         }
         static inline std::string getPropertyStr(const Property p) {
@@ -245,6 +253,8 @@ namespace acl {
             case PROP_POLICYTYPE:    return "policytype";
             case PROP_MAXQUEUESIZE:  return "maxqueuesize";
             case PROP_MAXQUEUECOUNT: return "maxqueuecount";
+            case PROP_MAXFILESIZE:   return "maxfilesize";
+            case PROP_MAXFILECOUNT:  return "maxfilecount";
             default: assert(false); // should never get here
             }
             return "";
@@ -266,6 +276,10 @@ namespace acl {
             if (str.compare("queuemaxsizeupperlimit")   == 0) return SPECPROP_MAXQUEUESIZEUPPERLIMIT;
             if (str.compare("queuemaxcountlowerlimit")  == 0) return SPECPROP_MAXQUEUECOUNTLOWERLIMIT;
             if (str.compare("queuemaxcountupperlimit")  == 0) return SPECPROP_MAXQUEUECOUNTUPPERLIMIT;
+            if (str.compare("filemaxsizelowerlimit")    == 0) return SPECPROP_MAXFILESIZELOWERLIMIT;
+            if (str.compare("filemaxsizeupperlimit")    == 0) return SPECPROP_MAXFILESIZEUPPERLIMIT;
+            if (str.compare("filemaxcountlowerlimit")   == 0) return SPECPROP_MAXFILECOUNTLOWERLIMIT;
+            if (str.compare("filemaxcountupperlimit")   == 0) return SPECPROP_MAXFILECOUNTUPPERLIMIT;
             // Allow old names in ACL file as aliases for newly-named properties
             if (str.compare("maxqueuesize")             == 0) return SPECPROP_MAXQUEUESIZEUPPERLIMIT;
             if (str.compare("maxqueuecount")            == 0) return SPECPROP_MAXQUEUECOUNTUPPERLIMIT;
@@ -289,6 +303,10 @@ namespace acl {
                 case SPECPROP_MAXQUEUESIZEUPPERLIMIT:  return "queuemaxsizeupperlimit";
                 case SPECPROP_MAXQUEUECOUNTLOWERLIMIT: return "queuemaxcountlowerlimit";
                 case SPECPROP_MAXQUEUECOUNTUPPERLIMIT: return "queuemaxcountupperlimit";
+                case SPECPROP_MAXFILESIZELOWERLIMIT:   return "filemaxsizelowerlimit";
+                case SPECPROP_MAXFILESIZEUPPERLIMIT:   return "filemaxsizeupperlimit";
+                case SPECPROP_MAXFILECOUNTLOWERLIMIT:  return "filemaxcountlowerlimit";
+                case SPECPROP_MAXFILECOUNTUPPERLIMIT:  return "filemaxcountupperlimit";
                 default: assert(false); // should never get here
             }
             return "";
diff --git a/cpp/src/qpid/broker/Broker.cpp b/cpp/src/qpid/broker/Broker.cpp
index c67d06f50c..979ad1c45e 100644
--- a/cpp/src/qpid/broker/Broker.cpp
+++ b/cpp/src/qpid/broker/Broker.cpp
@@ -1069,6 +1069,8 @@ std::pair<boost::shared_ptr<Queue>, bool> Broker::createQueue(
         params.insert(make_pair(acl::PROP_POLICYTYPE, settings.dropMessagesAtLimit ? "ring" : "reject"));
         params.insert(make_pair(acl::PROP_MAXQUEUECOUNT, boost::lexical_cast<string>(settings.maxDepth.getCount())));
         params.insert(make_pair(acl::PROP_MAXQUEUESIZE, boost::lexical_cast<string>(settings.maxDepth.getSize())));
+        params.insert(make_pair(acl::PROP_MAXFILECOUNT, boost::lexical_cast<string>(settings.maxFileCount)));
+        params.insert(make_pair(acl::PROP_MAXFILESIZE, boost::lexical_cast<string>(settings.maxFileSize)));
 
         if (!acl->authorise(userId,acl::ACT_CREATE,acl::OBJ_QUEUE,name,&params) )
             throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied queue create request from " << userId));
diff --git a/cpp/src/qpid/broker/QueueSettings.cpp b/cpp/src/qpid/broker/QueueSettings.cpp
index 91616636f1..b180a89e43 100644
--- a/cpp/src/qpid/broker/QueueSettings.cpp
+++ b/cpp/src/qpid/broker/QueueSettings.cpp
@@ -33,6 +33,8 @@ namespace broker {
 namespace {
 const std::string MAX_COUNT("qpid.max_count");
 const std::string MAX_SIZE("qpid.max_size");
+const std::string MAX_FILE_COUNT("qpid.file_count");
+const std::string MAX_FILE_SIZE("qpid.file_size");
 const std::string POLICY_TYPE("qpid.policy_type");
 const std::string POLICY_TYPE_REJECT("reject");
 const std::string POLICY_TYPE_RING("ring");
@@ -163,6 +165,12 @@ bool QueueSettings::handle(const std::string& key, const qpid::types::Variant& v
     } else if (key == ALERT_SIZE) {
         alertThreshold.setSize(value);
         return true;
+    } else if (key == MAX_FILE_COUNT && value.asUint64() > 0) {
+        maxFileCount = value.asUint64();
+        return false; // 'handle' here and also pass to store
+    } else if (key == MAX_FILE_SIZE && value.asUint64() > 0) {
+        maxFileSize = value.asUint64();
+        return false; // 'handle' here and also pass to store
     } else {
         return false;
     }
diff --git a/cpp/src/qpid/broker/QueueSettings.h b/cpp/src/qpid/broker/QueueSettings.h
index 2443624615..3256266df7 100644
--- a/cpp/src/qpid/broker/QueueSettings.h
+++ b/cpp/src/qpid/broker/QueueSettings.h
@@ -71,6 +71,10 @@ struct QueueSettings
     QueueDepth alertThreshold;
     int64_t alertRepeatInterval;
 
+    //file limits checked by Acl and shared with storeSettings
+    uint64_t maxFileSize;
+    uint64_t maxFileCount;
+
     //yuck, yuck
     qpid::framing::FieldTable storeSettings;
     std::map<std::string, qpid::types::Variant> original;
diff --git a/cpp/src/tests/acl.py b/cpp/src/tests/acl.py
index 570db20cf2..1020a2eff6 100755
--- a/cpp/src/tests/acl.py
+++ b/cpp/src/tests/acl.py
@@ -545,6 +545,123 @@ class ACLTests(TestBase010):
             self.fail(result)
 
 
+    def test_illegal_filemaxsize_upper_limit_spec(self):
+        """
+        Test illegal file policy
+        """
+        #
+        # Use filemaxsizeupperlimit
+        #
+        aclf = self.get_acl_file()
+        aclf.write('acl deny bob@QPID create queue name=q2 filemaxsizeupperlimit=-1\n')
+        aclf.write('acl allow all all')
+        aclf.close()
+
+        result = self.reload_acl()
+        expected = "-1 is not a valid value for 'filemaxsizeupperlimit', " \
+                   "values should be between 0 and 9223372036854775807";
+        if (result.find(expected) == -1):
+            self.fail(result)
+
+        aclf = self.get_acl_file()
+        aclf.write('acl deny bob@QPID create queue name=q2 filemaxsizeupperlimit=9223372036854775808\n')
+        aclf.write('acl allow all all')
+        aclf.close()
+
+        result = self.reload_acl()
+        expected = "9223372036854775808 is not a valid value for 'filemaxsizeupperlimit', " \
+                   "values should be between 0 and 9223372036854775807";
+        if (result.find(expected) == -1):
+            self.fail(result)
+
+
+
+    def test_illegal_filemaxcount_upper_limit_spec(self):
+        """
+        Test illegal file policy
+        """
+        #
+        # use maxfilecountupperlimit
+        #
+        aclf = self.get_acl_file()
+        aclf.write('acl deny bob@QPID create queue name=q2 filemaxcountupperlimit=-1\n')
+        aclf.write('acl allow all all')
+        aclf.close()
+
+        result = self.reload_acl()
+        expected = "-1 is not a valid value for 'filemaxcountupperlimit', " \
+                   "values should be between 0 and 9223372036854775807";
+        if (result.find(expected) == -1):
+            self.fail(result)
+
+        aclf = self.get_acl_file()
+        aclf.write('acl deny bob@QPID create queue name=q2 filemaxcountupperlimit=9223372036854775808\n')
+        aclf.write('acl allow all all')
+        aclf.close()
+
+        result = self.reload_acl()
+        expected = "9223372036854775808 is not a valid value for 'filemaxcountupperlimit', " \
+                   "values should be between 0 and 9223372036854775807";
+        if (result.find(expected) == -1):
+            self.fail(result)
+
+
+    def test_illegal_filemaxsize_lower_limit_spec(self):
+        """
+        Test illegal file policy
+        """
+        aclf = self.get_acl_file()
+        aclf.write('acl deny bob@QPID create queue name=q2 filemaxsizelowerlimit=-1\n')
+        aclf.write('acl allow all all')
+        aclf.close()
+
+        result = self.reload_acl()
+        expected = "-1 is not a valid value for 'filemaxsizelowerlimit', " \
+                   "values should be between 0 and 9223372036854775807";
+        if (result.find(expected) == -1):
+            self.fail(result)
+
+        aclf = self.get_acl_file()
+        aclf.write('acl deny bob@QPID create queue name=q2 filemaxsizelowerlimit=9223372036854775808\n')
+        aclf.write('acl allow all all')
+        aclf.close()
+
+        result = self.reload_acl()
+        expected = "9223372036854775808 is not a valid value for 'filemaxsizelowerlimit', " \
+                   "values should be between 0 and 9223372036854775807";
+        if (result.find(expected) == -1):
+            self.fail(result)
+
+
+
+    def test_illegal_filemaxcount_lower_limit_spec(self):
+        """
+        Test illegal file policy
+        """
+
+        aclf = self.get_acl_file()
+        aclf.write('acl deny bob@QPID create queue name=q2 filemaxcountlowerlimit=-1\n')
+        aclf.write('acl allow all all')
+        aclf.close()
+
+        result = self.reload_acl()
+        expected = "-1 is not a valid value for 'filemaxcountlowerlimit', " \
+                   "values should be between 0 and 9223372036854775807";
+        if (result.find(expected) == -1):
+            self.fail(result)
+
+        aclf = self.get_acl_file()
+        aclf.write('acl deny bob@QPID create queue name=q2 filemaxcountlowerlimit=9223372036854775808\n')
+        aclf.write('acl allow all all')
+        aclf.close()
+
+        result = self.reload_acl()
+        expected = "9223372036854775808 is not a valid value for 'filemaxcountlowerlimit', " \
+                   "values should be between 0 and 9223372036854775807";
+        if (result.find(expected) == -1):
+            self.fail(result)
+
+
    #=====================================
    # ACL queue tests
    #=====================================
@@ -834,6 +951,171 @@ class ACLTests(TestBase010):
                 self.fail("ACL should allow queue delete request for q4");
 
    #=====================================
+   # ACL file tests
+   #=====================================
+
+    def test_file_allow_mode(self):
+        """
+        Test cases for file acl in allow mode
+        """
+        aclf = self.get_acl_file()
+        aclf.write('acl deny bob@QPID access queue name=qf1\n')
+        aclf.write('acl deny bob@QPID create queue name=qf1 durable=true\n')
+        aclf.write('acl deny bob@QPID create queue name=qf2 exclusive=true policytype=ring\n')
+        aclf.write('acl deny bob@QPID access queue name=qf3\n')
+        aclf.write('acl deny bob@QPID purge queue name=qf3\n')
+        aclf.write('acl deny bob@QPID delete queue name=qf4\n')
+        aclf.write('acl deny bob@QPID create queue name=qf5 filemaxsizeupperlimit=1000 filemaxcountupperlimit=100\n')
+        aclf.write('acl allow all all')
+        aclf.close()
+
+        result = self.reload_acl()
+        if (result):
+            self.fail(result)
+
+        session = self.get_session('bob','bob')
+
+        try:
+            queue_options = {}
+            queue_options["qpid.file_count"] = 200
+            queue_options["qpid.file_size"] = 500
+            session.queue_declare(queue="qf5", exclusive=True, arguments=queue_options)
+            self.fail("ACL should deny queue create request with name=qf5, qpid.file_size=500 and qpid.file_count=200");
+        except qpid.session.SessionException, e:
+            self.assertEqual(403,e.args[0].error_code)
+            session = self.get_session('bob','bob')
+
+        try:
+            queue_options = {}
+            queue_options["qpid.file_count"] = 200
+            queue_options["qpid.file_size"] = 100
+            session.queue_declare(queue="qf2", exclusive=True, arguments=queue_options)
+        except qpid.session.SessionException, e:
+            if (403 == e.args[0].error_code):
+                self.fail("ACL should allow queue create request with name=qf2, qpid.file_size=100 and qpid.file_count=200 ");
+
+
+    def test_file_deny_mode(self):
+        """
+        Test cases for queue acl in deny mode
+        """
+        aclf = self.get_acl_file()
+        aclf.write('acl allow bob@QPID access queue name=qfd1\n')
+        aclf.write('acl allow bob@QPID create queue name=qfd1 durable=true\n')
+        aclf.write('acl allow bob@QPID create queue name=qfd2 exclusive=true policytype=ring\n')
+        aclf.write('acl allow bob@QPID access queue name=qfd3\n')
+        aclf.write('acl allow bob@QPID purge queue name=qfd3\n')
+        aclf.write('acl allow bob@QPID create queue name=qfd3\n')
+        aclf.write('acl allow bob@QPID create queue name=qfd4\n')
+        aclf.write('acl allow bob@QPID delete queue name=qfd4\n')
+        aclf.write('acl allow bob@QPID create queue name=qfd5 filemaxsizeupperlimit=1000 filemaxcountupperlimit=100\n')
+        aclf.write('acl allow bob@QPID create queue name=qfd6 filemaxsizelowerlimit=50 filemaxsizeupperlimit=100 filemaxcountlowerlimit=50 filemaxcountupperlimit=100\n')
+        aclf.write('acl allow anonymous all all\n')
+        aclf.write('acl deny all all')
+        aclf.close()
+
+        result = self.reload_acl()
+        if (result):
+            self.fail(result)
+
+        session = self.get_session('bob','bob')
+
+        try:
+            session.queue_declare(queue="qfd1", durable=True)
+        except qpid.session.SessionException, e:
+            if (403 == e.args[0].error_code):
+                self.fail("ACL should allow queue create request with name=qfd1 durable=true");
+
+        try:
+            session.queue_declare(queue="qfd1", durable=True, passive=True)
+        except qpid.session.SessionException, e:
+            if (403 == e.args[0].error_code):
+                self.fail("ACL should allow queue passive declare request with name=qfd1 durable=true passive=true");
+
+        try:
+            session.queue_declare(queue="qfd1", durable=False, passive=False)
+            self.fail("ACL should deny queue create request with name=qfd1 durable=true passive=false");
+        except qpid.session.SessionException, e:
+            self.assertEqual(403,e.args[0].error_code)
+            session = self.get_session('bob','bob')
+
+        try:
+            session.queue_declare(queue="qfd2", exclusive=False)
+            self.fail("ACL should deny queue create request with name=qfd2 exclusive=false");
+        except qpid.session.SessionException, e:
+            self.assertEqual(403,e.args[0].error_code)
+            session = self.get_session('bob','bob')
+
+        try:
+            queue_options = {}
+            queue_options["qpid.file_count"] = 200
+            queue_options["qpid.file_size"] = 500
+            session.queue_declare(queue="qfd5", arguments=queue_options)
+            self.fail("ACL should deny queue create request with name=qfd5 filemaxsizeupperlimit=500 filemaxcountupperlimit=200");
+        except qpid.session.SessionException, e:
+            self.assertEqual(403,e.args[0].error_code)
+            session = self.get_session('bob','bob')
+
+        try:
+            queue_options = {}
+            queue_options["qpid.file_count"] = 100
+            queue_options["qpid.file_size"] = 500
+            session.queue_declare(queue="qfd5", arguments=queue_options)
+        except qpid.session.SessionException, e:
+            if (403 == e.args[0].error_code):
+                self.fail("ACL should allow queue create request with name=qfd5 filemaxsizeupperlimit=500 filemaxcountupperlimit=200");
+
+        try:
+            queue_options = {}
+            queue_options["qpid.file_count"] = 49
+            queue_options["qpid.file_size"] = 100
+            session.queue_declare(queue="qfd6", arguments=queue_options)
+            self.fail("ACL should deny queue create request with name=qfd6 filemaxsizeupperlimit=100 filemaxcountupperlimit=49");
+        except qpid.session.SessionException, e:
+            self.assertEqual(403,e.args[0].error_code)
+            session = self.get_session('bob','bob')
+
+        try:
+            queue_options = {}
+            queue_options["qpid.file_count"] = 101
+            queue_options["qpid.file_size"] = 100
+            session.queue_declare(queue="qfd6", arguments=queue_options)
+            self.fail("ACL should allow queue create request with name=qfd6 filemaxsizeupperlimit=100 filemaxcountupperlimit=101");
+        except qpid.session.SessionException, e:
+            self.assertEqual(403,e.args[0].error_code)
+            session = self.get_session('bob','bob')
+
+        try:
+            queue_options = {}
+            queue_options["qpid.file_count"] = 100
+            queue_options["qpid.file_size"] = 49
+            session.queue_declare(queue="qfd6", arguments=queue_options)
+            self.fail("ACL should deny queue create request with name=qfd6 filemaxsizeupperlimit=49 filemaxcountupperlimit=100");
+        except qpid.session.SessionException, e:
+            self.assertEqual(403,e.args[0].error_code)
+            session = self.get_session('bob','bob')
+
+        try:
+            queue_options = {}
+            queue_options["qpid.file_count"] = 100
+            queue_options["qpid.file_size"] =101
+            session.queue_declare(queue="qfd6", arguments=queue_options)
+            self.fail("ACL should deny queue create request with name=qfd6 filemaxsizeupperlimit=101 filemaxcountupperlimit=100");
+        except qpid.session.SessionException, e:
+            self.assertEqual(403,e.args[0].error_code)
+            session = self.get_session('bob','bob')
+
+        try:
+            queue_options = {}
+            queue_options["qpid.file_count"] = 50
+            queue_options["qpid.file_size"] = 50
+            session.queue_declare(queue="qfd6", arguments=queue_options)
+        except qpid.session.SessionException, e:
+            if (403 == e.args[0].error_code):
+                self.fail("ACL should allow queue create request with name=qfd6 filemaxsizeupperlimit=50 filemaxcountupperlimit=50");
+
+
+   #=====================================
    # ACL exchange tests
    #=====================================