From 81b3e2ea21b42f5cae62792255029e87c641dc14 Mon Sep 17 00:00:00 2001
From: Jake Smith <jake.smith@lexisnexisrisk.com>
Date: Thu, 17 Oct 2024 18:45:06 +0100
Subject: [PATCH] HPCC-32822 Fix MP protocol error logged indefinitely in loop

The detection of a MP packet with an invalid header threw
an exception that was logged, but did not close the socket.
When the client closed the socket, the MP server was notified
but because it had nothing left to read of the [bad] header,
it saw the invalid protocol error again and rethrew the
same error.
The epoll handler loop continuously notified the handler
of the close event (because it did nothing with it), and
the protocol error was continuosly output.

The socket should be close on this and any other exception.

In case these events are too frequent, add a timer to log
the protocol errors less frequently, and log the header
bytes received to help diagnose what the source of these
'rogue' connections are.

Signed-off-by: Jake Smith <jake.smith@lexisnexisrisk.com>
---
 system/jlib/jdebug.hpp |  2 +-
 system/jlib/jmisc.cpp  |  2 +-
 system/mp/mpcomm.cpp   | 40 ++++++++++++++++++++++++++++------------
 3 files changed, 30 insertions(+), 14 deletions(-)

diff --git a/system/jlib/jdebug.hpp b/system/jlib/jdebug.hpp
index a4d5d41c705..e0983b892c5 100644
--- a/system/jlib/jdebug.hpp
+++ b/system/jlib/jdebug.hpp
@@ -275,7 +275,7 @@ class PeriodicTimer
 
 protected:
     cycle_t timePeriodCycles = 0;
-    cycle_t lastElapsedCycles = 0;
+    std::atomic<cycle_t> lastElapsedCycles{0};
 };
 
 
diff --git a/system/jlib/jmisc.cpp b/system/jlib/jmisc.cpp
index 9d1fbf5d504..f778d79cf2e 100644
--- a/system/jlib/jmisc.cpp
+++ b/system/jlib/jmisc.cpp
@@ -920,7 +920,7 @@ void throwExceptionIfAborting()
 
 StringBuffer & hexdump2string(byte const * in, size32_t inSize, StringBuffer & out)
 {
-    out.append("[");
+    out.appendf("%u bytes [", inSize);
     byte last = 0;
     unsigned seq = 1;
     for(unsigned i=0; i<inSize; ++i)
diff --git a/system/mp/mpcomm.cpp b/system/mp/mpcomm.cpp
index 6a3d17658ff..aa69fc8cfa4 100644
--- a/system/mp/mpcomm.cpp
+++ b/system/mp/mpcomm.cpp
@@ -221,10 +221,11 @@ struct MultiPacketHeader
 
 class DECL_EXCEPTION CMPException: public IMP_Exception, public CInterface
 {
+    StringAttr msg;
 public:
     IMPLEMENT_IINTERFACE;
 
-    CMPException(MessagePassingError err,const SocketEndpoint &ep) : error(err), endpoint(ep) 
+    CMPException(MessagePassingError err,const SocketEndpoint &ep, const char *_msg = nullptr) : error(err), endpoint(ep), msg(_msg)
     {
     }
 
@@ -240,6 +241,8 @@ class DECL_EXCEPTION CMPException: public IMP_Exception, public CInterface
         // change it from "MP link closed" to something more helpful
         case MPERR_link_closed:                 str.appendf("Unexpected process termination (ep:%s)",endpoint.getEndpointHostText(tmp).str()); break;
         }
+        if (msg.length())
+            str.append(" - ").append(msg);
         return str;
     }
     int             errorCode() const { return error; }
@@ -1813,6 +1816,8 @@ class ForwardPacketHandler // TAG_SYS_FORWARD
 };
 
 
+static PeriodicTimer periodicTimer(10*60*1000, false); // 10 minutes
+static std::atomic<__uint64> mpProtocolErrors{0};
 // --------------------------------------------------------
 
 class CMPPacketReader: public ISocketSelectNotify, public CInterface
@@ -1847,7 +1852,8 @@ class CMPPacketReader: public ISocketSelectNotify, public CInterface
     {
         if (!parent)
             return false;
-        bool gc = false; // if a gc is hit, then will fall through to close socket
+        bool closeSocket = false; // if a graceful close is hit, this will be set and will fall through to close socket
+        bool suppressException = false;
         try
         {
             while (true) // NB: breaks out if blocked (if (remaining) ..)
@@ -1872,7 +1878,7 @@ class CMPPacketReader: public ISocketSelectNotify, public CInterface
                 if (!gotPacketHdr)
                 {
                     CCycleTimer timer;
-                    gc = readtmsAllowClose(sock, activeptr, 0, remaining, szRead, timer.remainingMs(60000));
+                    closeSocket = readtmsAllowClose(sock, activeptr, 0, remaining, szRead, timer.remainingMs(60000));
                     remaining -= szRead;
                     activeptr += szRead;
                     if (remaining) // only possible if blocked.
@@ -1882,10 +1888,20 @@ class CMPPacketReader: public ISocketSelectNotify, public CInterface
                     if (hdr.version/0x100 != MP_PROTOCOL_VERSION/0x100)
                     {
                         // TBD IPV6 here
+                        mpProtocolErrors++;
                         SocketEndpoint ep;
                         sock->getPeerEndpoint(ep);
-                        IMP_Exception *e=new CMPException(MPERR_protocol_version_mismatch,ep);
-                        throw e;
+                        if (periodicTimer.hasElapsed())
+                        {
+                            VStringBuffer packetHdrBytes("[%" I64F "u incidents to date]. Packet Header: ", mpProtocolErrors.load());
+                            hexdump2string((byte const *)&hdr, sizeof(hdr), packetHdrBytes);
+                            throw new CMPException(MPERR_protocol_version_mismatch, ep, packetHdrBytes.str());
+                        }
+                        else
+                        {
+                            suppressException = true;
+                            throw new CMPException(MPERR_protocol_version_mismatch, ep);
+                        }
                     }
                     hdr.setMessageFields(*activemsg);
     #ifdef _FULLTRACE
@@ -1898,9 +1914,9 @@ class CMPPacketReader: public ISocketSelectNotify, public CInterface
                     gotPacketHdr = true;
                 }
 
-                if (!gc && remaining)
+                if (!closeSocket && remaining)
                 {
-                    gc = readtmsAllowClose(sock, activeptr, 0, remaining, szRead, WAIT_FOREVER);
+                    closeSocket = readtmsAllowClose(sock, activeptr, 0, remaining, szRead, WAIT_FOREVER);
                     remaining -= szRead;
                     activeptr += szRead;
                 }
@@ -1939,19 +1955,19 @@ class CMPPacketReader: public ISocketSelectNotify, public CInterface
                     }
                 }
                 while (activemsg);
-                if (gc)
+                if (closeSocket)
                     break;
             }
         }
         catch (IException *e)
         {
-            if (e->errorCode()!=JSOCKERR_graceful_close)
-                FLLOG(MCoperatorWarning, e,"MP(Packet Reader)");
+            if (!suppressException && e->errorCode()!=JSOCKERR_graceful_close)
+                FLLOG(MCoperatorWarning, e, "MP(Packet Reader)");
             e->Release();
-            gotPacketHdr = false;
+            closeSocket = true; // NB: this select handler will removed and not be notified again
         }
 
-        if (gc)
+        if (closeSocket)
         {
             // here due to error or graceful close, so close socket (ignore error as may be closed already)
             try