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

4 files changed, 61 insertions, 5 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; }
diff --git a/cpp/src/tests/cluster_test.cpp b/cpp/src/tests/cluster_test.cpp
index 72d5ee5a3f..dd4b34a2db 100644
--- a/cpp/src/tests/cluster_test.cpp
+++ b/cpp/src/tests/cluster_test.cpp
@@ -173,9 +173,49 @@ ConnectionSettings aclSettings(int port, const std::string& id) {
     return settings;
 }
 
+// An illegal frame body
+struct PoisonPill : public AMQBody {
+    virtual uint8_t type() const { return 0xFF; }
+    virtual void encode(Buffer& ) const {}
+    virtual void decode(Buffer& , uint32_t=0) {}
+    virtual uint32_t encodedSize() const { return 0; }
+
+    virtual void print(std::ostream&) const {};
+    virtual void accept(AMQBodyConstVisitor&) const {};
+
+    virtual AMQMethodBody* getMethod() { return 0; }
+    virtual const AMQMethodBody* getMethod() const { return 0; }
+
+    /** Match if same type and same class/method ID for methods */
+    static bool match(const AMQBody& , const AMQBody& ) { return false; }
+    virtual boost::intrusive_ptr<AMQBody> clone() const { return new PoisonPill; }
+};
+    
+QPID_AUTO_TEST_CASE(testBadClientData) {
+    // Ensure that bad data on a client connection closes the
+    // connection but does not stop the broker.
+    ClusterFixture::Args args;
+    prepareArgs(args, false);
+    args += "--log-enable=critical"; // Supress expected errors
+    ClusterFixture cluster(2, args, -1);
+    Client c0(cluster[0]);
+    Client c1(cluster[1]);
+    boost::shared_ptr<client::ConnectionImpl> ci =
+        client::ConnectionAccess::getImpl(c0.connection);
+    AMQFrame poison(boost::intrusive_ptr<AMQBody>(new PoisonPill));
+    ci->handle(poison);
+    {
+        ScopedSuppressLogging sl;
+        BOOST_CHECK_THROW(c0.session.queueQuery("q"), TransportFailure);
+    }
+    Client c00(cluster[0]);
+    BOOST_CHECK_EQUAL(c00.session.queueQuery("q").getQueue(), "");
+    BOOST_CHECK_EQUAL(c1.session.queueQuery("q").getQueue(), "");
+}
+
 #if 0
-// FIXME aconway 2009-03-10: This test passes but exposes a memory leak in the SASL client code.
-// Enable it when the leak is fixed.
+// FIXME aconway 2009-03-10: This test passes but exposes a memory
+// leak in the SASL client code.  Enable it when the leak is fixed.
 
 QPID_AUTO_TEST_CASE(testAcl) {
     ofstream policyFile("cluster_test.acl");