Handle invalid AMPQ data to a cluster by closing the offending connection.

Prior to this fix, invalid data shut down the whole cluster.


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

3 files changed, 19 insertions, 3 deletions

diff --git a/cpp/src/qpid/cluster/Cluster.cpp b/cpp/src/qpid/cluster/Cluster.cpp
index fe6958244f..a472287a35 100644
--- a/cpp/src/qpid/cluster/Cluster.cpp
+++ b/cpp/src/qpid/cluster/Cluster.cpp
@@ -103,6 +103,7 @@
 #include "qpid/framing/AllInvoker.h"
 #include "qpid/framing/ClusterConfigChangeBody.h"
 #include "qpid/framing/ClusterConnectionDeliverCloseBody.h"
+#include "qpid/framing/ClusterConnectionAbortBody.h"
 #include "qpid/framing/ClusterConnectionDeliverDoOutputBody.h"
 #include "qpid/framing/ClusterErrorCheckBody.h"
 #include "qpid/framing/ClusterReadyBody.h"
@@ -245,6 +246,7 @@ void Cluster::erase(const ConnectionId& id) {
 
 // Called by Connection::deliverClose() in deliverFrameQueue thread.
 void Cluster::erase(const ConnectionId& id, Lock&) {
+    QPID_LOG(debug, *this << " erasing connection " << id);
     connections.erase(id);
     decoder.erase(id);
 }
@@ -334,8 +336,16 @@ void Cluster::deliveredEvent(const Event& e) {
     else if(!discarding) {    
         if (e.isControl())
             deliverFrame(EventFrame(e, e.getFrame()));
-        else
-            decoder.decode(e, e.getData());
+        else {
+            try { decoder.decode(e, e.getData()); }
+            catch (const Exception& ex) {
+                // Close a connection that is sending us invalid data.
+                QPID_LOG(error, *this << " aborting connection "
+                         << e.getConnectionId() << ": " << ex.what());
+                framing::AMQFrame abort((ClusterConnectionAbortBody()));
+                deliverFrame(EventFrame(EventHeader(CONTROL, e.getConnectionId()), abort));
+            }
+        }
     }
     else // Discard connection events if discarding is set.
         QPID_LOG(trace, *this << " DROP: " << e);
diff --git a/cpp/src/qpid/cluster/Connection.cpp b/cpp/src/qpid/cluster/Connection.cpp
index e7dac82159..2db8879eb5 100644
--- a/cpp/src/qpid/cluster/Connection.cpp
+++ b/cpp/src/qpid/cluster/Connection.cpp
@@ -198,6 +198,12 @@ void Connection::deliverClose () {
     cluster.erase(self);
 }
 
+// The connection has been killed for misbehaving
+void Connection::abort() {
+    connection.abort();
+    cluster.erase(self);
+}
+
 // Member of a shadow connection left the cluster.
 void Connection::left() {
     assert(isShadow());
diff --git a/cpp/src/qpid/cluster/Connection.h b/cpp/src/qpid/cluster/Connection.h
index 51aab92bfc..0b7c151e8a 100644
--- a/cpp/src/qpid/cluster/Connection.h
+++ b/cpp/src/qpid/cluster/Connection.h
@@ -150,7 +150,7 @@ class Connection :
     void exchange(const std::string& encoded);
 
     void giveReadCredit(int credit);
-
+    void abort();
     void deliverClose();
 
     OutputInterceptor& getOutput() { return output; }