HPCC-24445 Validate sort connections

Signed-off-by: Jake Smith <jake.smith@lexisnexisrisk.com>

thorlcr/msort/tsortl.cpp

@@ -65,14 +65,22 @@ struct TransferStreamHeader
 {
     rowcount_t numrecs;
     rowcount_t pos;
-    size32_t recsize;
     unsigned id;
-    TransferStreamHeader(rowcount_t _pos, rowcount_t _numrecs, unsigned _recsize, unsigned _id)
-        : pos(_pos), numrecs(_numrecs), recsize(_recsize), id(_id)
+    unsigned crc = 0;
+    TransferStreamHeader(rowcount_t _pos, rowcount_t _numrecs, unsigned _id)
+        : pos(_pos), numrecs(_numrecs), id(_id)
     {
+        crc = getCrc();
     }
     TransferStreamHeader() {}
-    void winrev() { _WINREV(pos); _WINREV(numrecs); _WINREV(recsize);  _WINREV(id); }
+    void winrev() { _WINREV(pos); _WINREV(numrecs); _WINREV(id); _WINREV(crc); }
+    unsigned getCrc() const
+    {
+        unsigned retCrc = crc32((const char *)&numrecs, sizeof(numrecs), 0);
+        retCrc = crc32((const char *)&pos, sizeof(pos), retCrc);
+        retCrc = crc32((const char *)&id, sizeof(id), retCrc);
+        return retCrc;
+    }
 };
 
 
@@ -281,7 +289,7 @@ public:
 IRowStream *ConnectMergeRead(unsigned id, IThorRowInterfaces *rowif,SocketEndpoint &nodeaddr,rowcount_t startrec,rowcount_t numrecs)
 {
     Owned<ISocket> socket = DoConnect(nodeaddr);
-    TransferStreamHeader hdr(startrec,numrecs,0,id);
+    TransferStreamHeader hdr(startrec, numrecs, id);
 #ifdef _FULL_TRACE
     StringBuffer s;
     nodeaddr.getUrlStr(s);
@@ -296,8 +304,30 @@ IRowStream *ConnectMergeRead(unsigned id, IThorRowInterfaces *rowif,SocketEndpoi
 ISocketRowWriter *ConnectMergeWrite(IThorRowInterfaces *rowif,ISocket *socket,size32_t bufsize,rowcount_t &startrec,rowcount_t &numrecs)
 {
     TransferStreamHeader hdr;
-    socket->read(&hdr,sizeof(hdr));
+    unsigned remaining = sizeof(hdr);
+    byte *dst = (byte *)&hdr;
+
+    /*
+     * A client has connected at this stage, the hdr should be sent swiftly.
+     * A generous 1 minute timeout between reads, if longer, timeout exception,
+     * will be thrown, and the connection ignored.
+     */
+    while (true)
+    {
+        size32_t read;
+        socket->readtms(dst, 1, remaining, read, 60*1000); // 1 min timeout
+        if (read == remaining)
+            break;
+        remaining -= read;
+        dst += read;
+    }
     hdr.winrev();
+    if (hdr.getCrc() != hdr.crc)
+    {
+        char name[100];
+        int port = socket->peer_name(name,sizeof(name));
+        throw makeStringExceptionV(TE_InvalidSortConnect, "Invalid SORT connection from: %s:%u", name, port);
+    }
     startrec = hdr.pos;
     numrecs = hdr.numrecs;
 #ifdef _FULL_TRACE

thorlcr/msort/tsorts1.cpp

@@ -336,28 +336,38 @@ public:
         DBGLOG("CSortTransferServerThread started port %d",slave.getTransferPort());
         unsigned numretries = 10;
         try {
-            while (!term) {
-                ISocket* socket = server->accept(true);
-                if (!socket) {
+            while (!term)
+            {
+                Owned<ISocket> socket = server->accept(true);
+                if (!socket)
                     break;
-                }
 
                 rowcount_t poscount=0;
                 rowcount_t numrecs=0;
                 ISocketRowWriter *strm=NULL;
-                try {
+                try
+                {
                     waitRowIF();
                     strm = ConnectMergeWrite(rowif,socket,0x100000,poscount,numrecs);
                 }
-                catch (IJSOCK_Exception *e) { // retry if failed
-                    PrintExceptionLog(e,"WARNING: Exception(ConnectMergeWrite)");
+                catch (IJSOCK_Exception *e) // retry if failed
+                {
+                    PrintExceptionLog(e, "WARNING: Exception(ConnectMergeWrite)");
                     if (--numretries==0)
-                        throw e;
+                        throw;
+                    e->Release();
+                    continue;
+                }
+                catch (IException *e) // only retry if serialization check failed, indicating possible foreign client connect
+                {
+                    PrintExceptionLog(e, "WARNING: Exception(ConnectMergeWrite)");
+                    if (TE_InvalidSortConnect != e->errorCode() || (--numretries==0))
+                        throw;
                     e->Release();
                     continue;
                 }
                 CriticalBlock block(childsect);
-                add(strm,socket,poscount,numrecs);
+                add(strm, socket.getClear(), poscount, numrecs);
             }
         }
         catch (IJSOCK_Exception *e)
@@ -414,7 +424,7 @@ public:
             selecthandler.setown(createSocketSelectHandler("SORT"));
             selecthandler->start();
         }
-        CSortMerge *sub = new CSortMerge(this,socket,strm,poscount,numrecs,selecthandler);
+        CSortMerge *sub = new CSortMerge(this,socket,strm,poscount,numrecs,selecthandler); // NB: takes ownership of 'socket'
         children.append(*sub);
         selecthandler->add(socket,SELECTMODE_READ,sub);
 

thorlcr/shared/thexception.hpp

@@ -163,7 +163,8 @@
 #define TE_MissingOptionalFile                  TE_Base + 140
 #define TE_UnsupportedSortOrder                 TE_Base + 141
 #define TE_CostExceeded                         TE_Base + 142
-#define TE_Final                                TE_Base + 143       // keep this last
+#define TE_InvalidSortConnect                   TE_Base + 143
+#define TE_Final                                TE_Base + 144       // keep this last
 #define ISTHOREXCEPTION(n) (n > TE_Base && n < TE_Final)
 
 #endif