from (aconway)... added facility to disable management methods.

Use this facility to disable several methods that are unsafe for clusters when running in a cluster.


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

3 files changed, 39 insertions, 0 deletions

diff --git a/cpp/src/qpid/cluster/ClusterPlugin.cpp b/cpp/src/qpid/cluster/ClusterPlugin.cpp
index 56c50eafae..1bebbe7a43 100644
--- a/cpp/src/qpid/cluster/ClusterPlugin.cpp
+++ b/cpp/src/qpid/cluster/ClusterPlugin.cpp
@@ -146,8 +146,21 @@ struct ClusterPlugin : public Plugin {
         }
     }
 
+    void disallow(ManagementAgent* agent, const string& className, const string& methodName) {
+        string message = "Management method " + className + ":" + methodName + " is not allowed on a clustered broker.";
+        agent->disallow(className, methodName, message);
+    }
+    void disallowManagementMethods(ManagementAgent* agent) {
+        if (!agent) return;
+        disallow(agent, "queue", "purge");
+        disallow(agent, "session", "detach");
+        disallow(agent, "session", "close");
+        disallow(agent, "connection", "close");
+    }
+
     void initialize(Plugin::Target& target) {
         Broker* broker = dynamic_cast<Broker*>(&target);
+        disallowManagementMethods(broker->getManagementAgent());
         if (broker && cluster)
             cluster->initialize();
     }
diff --git a/cpp/src/qpid/management/ManagementAgent.cpp b/cpp/src/qpid/management/ManagementAgent.cpp
index 8dce82ba84..4998b274e8 100644
--- a/cpp/src/qpid/management/ManagementAgent.cpp
+++ b/cpp/src/qpid/management/ManagementAgent.cpp
@@ -459,6 +459,16 @@ void ManagementAgent::handleMethodRequestLH (Buffer& inBuffer, string replyToKey
     inBuffer.getShortString(methodName);
     encodeHeader(outBuffer, 'm', sequence);
 
+    DisallowedMethods::const_iterator i = disallowed.find(std::make_pair(className, methodName));
+    if (i != disallowed.end()) {
+        outBuffer.putLong(Manageable::STATUS_FORBIDDEN);
+        outBuffer.putMediumString(i->second);
+        outLen = MA_BUFFER_SIZE - outBuffer.available();
+        outBuffer.reset();
+        sendBuffer(outBuffer, outLen, dExchange, replyToKey);
+        return;
+    }
+
     if (acl != 0) {
         string userId = ((const qpid::broker::ConnectionState*) connToken)->getUserId();
         map<acl::Property, string> params;
@@ -1133,3 +1143,7 @@ uint64_t ManagementAgent::allocateId(Manageable* object)
     if (allocator.get()) return allocator->getIdFor(object);
     return 0;
 }
+
+void ManagementAgent::disallow(const std::string& className, const std::string& methodName, const std::string& message) {
+    disallowed[std::make_pair(className, methodName)] = message;
+}
diff --git a/cpp/src/qpid/management/ManagementAgent.h b/cpp/src/qpid/management/ManagementAgent.h
index 2411e6c277..34d53f778d 100644
--- a/cpp/src/qpid/management/ManagementAgent.h
+++ b/cpp/src/qpid/management/ManagementAgent.h
@@ -34,6 +34,8 @@
 #include "qmf/org/apache/qpid/broker/Agent.h"
 #include <qpid/framing/AMQFrame.h>
 #include <memory>
+#include <string>
+#include <map>
 
 namespace qpid {
 namespace management {
@@ -59,6 +61,7 @@ public:
     SEV_DEFAULT = 8
     } severity_t;
 
+
     ManagementAgent ();
     virtual ~ManagementAgent ();
 
@@ -90,6 +93,10 @@ public:
 
     void setAllocator(std::auto_ptr<IdAllocator> allocator);
     uint64_t allocateId(Manageable* object);
+
+    /** Disallow a method. Attempts to call it will receive an exception with message. */
+    void disallow(const std::string& className, const std::string& methodName, const std::string& message);
+                  
 private:
     struct Periodic : public qpid::broker::TimerTask
     {
@@ -192,6 +199,11 @@ private:
 
     std::auto_ptr<IdAllocator> allocator;
 
+    typedef std::pair<std::string,std::string> MethodName;
+    typedef std::map<MethodName, std::string> DisallowedMethods;
+    DisallowedMethods disallowed;
+
+
 #   define MA_BUFFER_SIZE 65536
     char inputBuffer[MA_BUFFER_SIZE];
     char outputBuffer[MA_BUFFER_SIZE];