Bugfix: Fix interrupted network read operation (#120)

### Problem

The `MQTT_ProcessLoop` and `MQTT_ReceiveLoop` read incoming MQTT packet payload over the network by calling the `recvExact` function. The `recvExact` function can be called multiple times to read the expected number of bytes for the MQTT packet but it also implements a timeout functionality of receiving the expected number of payload within the timeout value passed to the function.
This causes problems when the `Transport_Recv` call returns less than requested number of bytes, and there is a timeout (for example, when calling `MQTT_ProcessLoop` with 0ms duration) which causes the function to assume failure instead of reading the remaining payload of the MQTT packet by calling `Transport_Recv` again. Thus, in such cases, the MQTT connection is severed prematurely even though there is a high probability of receiving the remaining bytes of the MQTT packet over the network.

### Solution
Instead of implementing a timeout on the entire duration of receiving the expected number of remaining MQTT packet bytes in `recvExact`, the use of timeout is being changed to be relevant only on the total time of receiving 0 bytes over the network over multiple calls to `Transport_Recv`.
As this modified meaning of the timeout duration is now unrelated to the timeout duration that the `MQTT_ProcessLoop` or `MQTT_ReceiveLoop` functions are called, a new configuration constant for the `recvExact` timeout value, `MQTT_RECV_POLLING_TIMEOUT_MS`, has been added to the library which will carry a default value of 10ms. 

Co-authored-by: Sarena Meas <[email protected]>

Author: aggarw13

docs/doxygen/pages.dox

@@ -188,6 +188,9 @@ Some configuration settings are C pre-processor constants, and some are function
 @section MQTT_PINGRESP_TIMEOUT_MS
 @copydoc MQTT_PINGRESP_TIMEOUT_MS
 
+@section MQTT_RECV_POLLING_TIMEOUT_MS
+@copydoc MQTT_RECV_POLLING_TIMEOUT_MS
+
 @section MQTT_MAX_CONNACK_RECEIVE_RETRY_COUNT
 @copydoc MQTT_MAX_CONNACK_RECEIVE_RETRY_COUNT
 

lexicon.txt

@@ -174,6 +174,7 @@ mytcpsocketcontext
 mytlscontext
 networkbuffer
 networkcontext
+networkinterfacereceivestub
 networkinterfacesendstub
 networkrecv
 networksend
@@ -279,6 +280,7 @@ receivepacket
 recordcount
 recordindex
 recv
+recvexact
 recvfunc
 reestablishment
 remaininglength

source/core_mqtt.c

@@ -71,17 +71,25 @@ static uint32_t calculateElapsedTime( uint32_t later,
 static MQTTPubAckType_t getAckFromPacketType( uint8_t packetType );
 
 /**
- * @brief Receive bytes into the network buffer, with a timeout.
+ * @brief Receive bytes into the network buffer.
  *
  * @param[in] pContext Initialized MQTT Context.
  * @param[in] bytesToRecv Number of bytes to receive.
- * @param[in] timeoutMs Time remaining to receive the packet.
+ *
+ * @note This operation calls the transport receive function
+ * repeatedly to read bytes from the network until either:
+ * 1. The requested number of bytes @a bytesToRecv are read.
+ *                    OR
+ * 2. No data is received from the network for MQTT_RECV_POLLING_TIMEOUT_MS duration.
+ *
+ *                    OR
+ * 3. There is an error in reading from the network.
+ *
  *
  * @return Number of bytes received, or negative number on network error.
  */
 static int32_t recvExact( const MQTTContext_t * pContext,
-                          size_t bytesToRecv,
-                          uint32_t timeoutMs );
+                          size_t bytesToRecv );
 
 /**
  * @brief Discard a packet from the transport interface.
@@ -683,13 +691,12 @@ static MQTTPubAckType_t getAckFromPacketType( uint8_t packetType )
 /*-----------------------------------------------------------*/
 
 static int32_t recvExact( const MQTTContext_t * pContext,
-                          size_t bytesToRecv,
-                          uint32_t timeoutMs )
+                          size_t bytesToRecv )
 {
     uint8_t * pIndex = NULL;
     size_t bytesRemaining = bytesToRecv;
     int32_t totalBytesRecvd = 0, bytesRecvd;
-    uint32_t entryTimeMs = 0U, elapsedTimeMs = 0U;
+    uint32_t lastDataRecvTimeMs = 0U, timeSinceLastRecvMs = 0U;
     TransportRecv_t recvFunc = NULL;
     MQTTGetCurrentTimeFunc_t getTimeStampMs = NULL;
     bool receiveError = false;
@@ -704,7 +711,8 @@ static int32_t recvExact( const MQTTContext_t * pContext,
     recvFunc = pContext->transportInterface.recv;
     getTimeStampMs = pContext->getTime;
 
-    entryTimeMs = getTimeStampMs();
+    /* Part of the MQTT packet has been read before calling this function. */
+    lastDataRecvTimeMs = getTimeStampMs();
 
     while( ( bytesRemaining > 0U ) && ( receiveError == false ) )
     {
@@ -719,8 +727,11 @@ static int32_t recvExact( const MQTTContext_t * pContext,
             totalBytesRecvd = bytesRecvd;
             receiveError = true;
         }
-        else
+        else if( bytesRecvd > 0 )
         {
+            /* Reset the starting time as we have received some data from the network. */
+            lastDataRecvTimeMs = getTimeStampMs();
+
             /* It is a bug in the application's transport receive implementation
              * if more bytes than expected are received. To avoid a possible
              * overflow in converting bytesRemaining from unsigned to signed,
@@ -732,18 +743,20 @@ static int32_t recvExact( const MQTTContext_t * pContext,
             totalBytesRecvd += ( int32_t ) bytesRecvd;
             pIndex += bytesRecvd;
             LogDebug( ( "BytesReceived=%ld, BytesRemaining=%lu, "
-                        "TotalBytesReceived=%ld.",
                         ( long int ) bytesRecvd,
-                        ( unsigned long ) bytesRemaining,
-                        ( long int ) totalBytesRecvd ) );
+                        ( unsigned long ) bytesRemaining ) );
         }
-
-        elapsedTimeMs = calculateElapsedTime( getTimeStampMs(), entryTimeMs );
-
-        if( ( bytesRemaining > 0U ) && ( elapsedTimeMs >= timeoutMs ) )
+        else
         {
-            LogError( ( "Time expired while receiving packet." ) );
-            receiveError = true;
+            /* No bytes were read from the network. */
+            timeSinceLastRecvMs = calculateElapsedTime( getTimeStampMs(), lastDataRecvTimeMs );
+
+            /* Check for timeout if we have been waiting to receive any byte on the network. */
+            if( timeSinceLastRecvMs >= MQTT_RECV_POLLING_TIMEOUT_MS )
+            {
+                LogError( ( "Time expired while receiving packet." ) );
+                receiveError = true;
+            }
         }
     }
 
@@ -760,7 +773,6 @@ static MQTTStatus_t discardPacket( const MQTTContext_t * pContext,
     int32_t bytesReceived = 0;
     size_t bytesToReceive = 0U;
     uint32_t totalBytesReceived = 0U, entryTimeMs = 0U, elapsedTimeMs = 0U;
-    uint32_t remainingTimeMs = timeoutMs;
     MQTTGetCurrentTimeFunc_t getTimeStampMs = NULL;
     bool receiveError = false;
 
@@ -779,7 +791,7 @@ static MQTTStatus_t discardPacket( const MQTTContext_t * pContext,
             bytesToReceive = remainingLength - totalBytesReceived;
         }
 
-        bytesReceived = recvExact( pContext, bytesToReceive, remainingTimeMs );
+        bytesReceived = recvExact( pContext, bytesToReceive );
 
         if( bytesReceived != ( int32_t ) bytesToReceive )
         {
@@ -795,12 +807,8 @@ static MQTTStatus_t discardPacket( const MQTTContext_t * pContext,
 
             elapsedTimeMs = calculateElapsedTime( getTimeStampMs(), entryTimeMs );
 
-            /* Update remaining time and check for timeout. */
-            if( elapsedTimeMs < timeoutMs )
-            {
-                remainingTimeMs = timeoutMs - elapsedTimeMs;
-            }
-            else
+            /* Check for timeout. */
+            if( elapsedTimeMs >= timeoutMs )
             {
                 LogError( ( "Time expired while discarding packet." ) );
                 receiveError = true;
@@ -846,7 +854,7 @@ static MQTTStatus_t receivePacket( const MQTTContext_t * pContext,
     else
     {
         bytesToReceive = incomingPacket.remainingLength;
-        bytesReceived = recvExact( pContext, bytesToReceive, remainingTimeMs );
+        bytesReceived = recvExact( pContext, bytesToReceive );
 
         if( bytesReceived == ( int32_t ) bytesToReceive )
         {
@@ -2167,7 +2175,7 @@ MQTTStatus_t MQTT_ProcessLoop( MQTTContext_t * pContext,
             elapsedTimeMs = calculateElapsedTime( pContext->getTime(),
                                                   entryTimeMs );
 
-            if( elapsedTimeMs > timeoutMs )
+            if( elapsedTimeMs >= timeoutMs )
             {
                 break;
             }

source/include/core_mqtt.h

@@ -603,6 +603,14 @@ MQTTStatus_t MQTT_Disconnect( MQTTContext_t * pContext );
  * @param[in] timeoutMs Minimum time in milliseconds that the receive loop will
  * run, unless an error occurs.
  *
+ * @note Calling this function blocks the calling context for a time period that
+ * depends on the passed @p timeoutMs, the configuration macro, #MQTT_RECV_POLLING_TIMEOUT_MS,
+ * and the underlying transport interface implementation timeouts, unless an error
+ * occurs.
+ *    Blocking Time = Max( timeoutMs parameter,
+ *                         MQTT_RECV_POLLING_TIMEOUT_MS,
+ *                         Transport interface send/recv implementation timeout )
+ *
  * @return #MQTTBadParameter if context is NULL;
  * #MQTTRecvFailed if a network error occurs during reception;
  * #MQTTSendFailed if a network error occurs while sending an ACK or PINGREQ;
@@ -655,6 +663,14 @@ MQTTStatus_t MQTT_ProcessLoop( MQTTContext_t * pContext,
  * @param[in] timeoutMs Minimum time in milliseconds that the receive loop will
  * run, unless an error occurs.
  *
+ * @note Calling this function blocks the calling context for a time period that
+ * depends on the passed @p timeoutMs, the configuration macro, #MQTT_RECV_POLLING_TIMEOUT_MS,
+ * and the underlying transport interface implementation timeouts, unless an error
+ * occurs.
+ *    Blocking Time = Max( timeoutMs parameter,
+ *                         MQTT_RECV_POLLING_TIMEOUT_MS,
+ *                         Transport interface send/recv implementation timeout )
+ *
  * @return #MQTTBadParameter if context is NULL;
  * #MQTTRecvFailed if a network error occurs during reception;
  * #MQTTSendFailed if a network error occurs while sending an ACK or PINGREQ;

source/include/core_mqtt_config_defaults.h

@@ -108,6 +108,27 @@
     #define MQTT_PINGRESP_TIMEOUT_MS    ( 500U )
 #endif
 
+/**
+ * @brief The maximum duration between non-empty network reads while
+ * receiving an MQTT packet via the #MQTT_ProcessLoop or #MQTT_ReceiveLoop
+ * API functions.
+ *
+ * When an incoming MQTT packet is detected, the transport receive function
+ * may be called multiple times until all of the expected number of bytes of the
+ * packet are received. This timeout represents the maximum polling duration that
+ * is allowed without any data reception from the network for the incoming packet.
+ * If the timeout expires, the #MQTT_ProcessLoop and #MQTT_ReceiveLoop functions
+ * return #MQTTRecvFailed.
+ *
+ * <b>Possible values:</b> Any positive integer up to SIZE_MAX. Recommended to
+ * use a small timeout value. <br>
+ * <b>Default value:</b> `10`
+ *
+ */
+#ifndef MQTT_RECV_POLLING_TIMEOUT_MS
+    #define MQTT_RECV_POLLING_TIMEOUT_MS    ( 10U )
+#endif
+
 /**
  * @brief Macro that is called in the MQTT library for logging "Error" level
  * messages.

source/interface/transport_interface.h

@@ -163,7 +163,11 @@ typedef struct NetworkContext NetworkContext_t;
  * @param[in] pBuffer Buffer to receive the data into.
  * @param[in] bytesToRecv Number of bytes requested from the network.
  *
- * @return The number of bytes received or a negative error code.
+ * @return The number of bytes received or a negative value to indicate
+ * error.
+ * @note If no data is available on the network to read and no error
+ * has occurred, zero MUST be the return value. Zero MUST NOT be used
+ * if a network disconnection has occurred.
  */
 /* @[define_transportrecv] */
 typedef int32_t ( * TransportRecv_t )( NetworkContext_t * pNetworkContext,

test/cbmc/include/core_mqtt_config.h

@@ -70,4 +70,22 @@ struct NetworkContext
  */
 #define MQTT_PINGRESP_TIMEOUT_MS                ( 500U )
 
+/**
+ * @brief The maximum duration of receiving no data over network when
+ * attempting to read an incoming MQTT packet by the #MQTT_ProcessLoop or
+ * #MQTT_ReceiveLoop API functions.
+ *
+ * When an incoming MQTT packet is detected, the transport receive function
+ * may be called multiple times until all the expected number of bytes for the
+ * packet are received. This timeout represents the maximum duration of polling
+ * for any data to be received over the network for the incoming.
+ * If the timeout expires, the #MQTT_ProcessLoop or #MQTT_ReceiveLoop functions
+ * return #MQTTRecvFailed.
+ *
+ * This is set to 1 to exit right away after a zero is received in the transport
+ * receive stub. There is no added value, in proving memory safety, to repeat
+ * the logic that checks if the polling timeout is reached.
+ */
+#define MQTT_RECV_POLLING_TIMEOUT_MS            ( 1U )
+
 #endif /* ifndef CORE_MQTT_CONFIG_H_ */

test/cbmc/proofs/MQTT_Connect/Makefile

@@ -32,12 +32,17 @@ MAX_NETWORK_SEND_TRIES=3
 # time out of 3 we can get coverage of the entire function. Another iteration
 # performed will unnecessarily duplicate the proof.
 MQTT_RECEIVE_TIMEOUT=3
+# The NetworkInterfaceReceiveStub is called once for getting the incoming packet
+# type with one byte of data, then it is called multiple times to reveive the
+# packet.
+MAX_NETWORK_RECV_TRIES=4
 # Please see test/cbmc/include/core_mqtt_config.h for more
 # information on these defines.
 MQTT_STATE_ARRAY_MAX_COUNT=11
 MQTT_MAX_CONNACK_RECEIVE_RETRY_COUNT=3
 DEFINES += -DMQTT_RECEIVE_TIMEOUT=$(MQTT_RECEIVE_TIMEOUT)
 DEFINES += -DMAX_NETWORK_SEND_TRIES=$(MAX_NETWORK_SEND_TRIES)
+DEFINES += -DMAX_NETWORK_RECV_TRIES=$(MAX_NETWORK_RECV_TRIES)
 INCLUDES +=
 
 # These functions do not coincide with the call graph of MQTT_Connect, but are
@@ -54,10 +59,10 @@ REMOVE_FUNCTION_BODY += __CPROVER_file_local_core_mqtt_c_handleKeepAlive
 # function.
 REMOVE_FUNCTION_BODY += memcpy
 
-# The loops below are unwound once more than the timeout. The loops below use
+# The loop below is unwound once more than the timeout. The loop below uses
 # the user passed in timeout to break the loop.
-UNWINDSET += __CPROVER_file_local_core_mqtt_c_recvExact.0:$(MQTT_RECEIVE_TIMEOUT)
 UNWINDSET += __CPROVER_file_local_core_mqtt_c_discardPacket.0:$(MQTT_RECEIVE_TIMEOUT)
+UNWINDSET += __CPROVER_file_local_core_mqtt_c_recvExact.0:$(MAX_NETWORK_RECV_TRIES)
 # If the user passed in timeout is zero, then the loop will run until the
 # MQTT_MAX_CONNACK_RECEIVE_RETRY_COUNT is reached.
 UNWINDSET += __CPROVER_file_local_core_mqtt_c_receiveConnack.0:$(MQTT_MAX_CONNACK_RECEIVE_RETRY_COUNT)

test/cbmc/proofs/MQTT_ProcessLoop/Makefile

@@ -32,11 +32,16 @@ MQTT_RECEIVE_TIMEOUT=3
 # Please see test/cbmc/stubs/network_interface_subs.c for
 # more information on MAX_NETWORK_SEND_TRIES.
 MAX_NETWORK_SEND_TRIES=3
+# The NetworkInterfaceReceiveStub is called once for getting the incoming packet
+# type with one byte of data, then it is called multiple times to reveive the
+# packet.
+MAX_NETWORK_RECV_TRIES=4
 # Please see test/cbmc/include/core_mqtt_config.h for more
 # information.
 MQTT_STATE_ARRAY_MAX_COUNT=11
 DEFINES += -DMQTT_RECEIVE_TIMEOUT=$(MQTT_RECEIVE_TIMEOUT)
 DEFINES += -DMAX_NETWORK_SEND_TRIES=$(MAX_NETWORK_SEND_TRIES)
+DEFINES += -DMAX_NETWORK_RECV_TRIES=$(MAX_NETWORK_RECV_TRIES)
 INCLUDES +=
 
 # These functions have their memory saftey proven in other harnesses.
@@ -46,7 +51,7 @@ REMOVE_FUNCTION_BODY += MQTT_SerializeAck
 
 UNWINDSET += MQTT_ProcessLoop.0:$(MQTT_RECEIVE_TIMEOUT)
 UNWINDSET += __CPROVER_file_local_core_mqtt_c_discardPacket.0:$(MQTT_RECEIVE_TIMEOUT)
-UNWINDSET += __CPROVER_file_local_core_mqtt_c_recvExact.0:$(MQTT_RECEIVE_TIMEOUT)
+UNWINDSET += __CPROVER_file_local_core_mqtt_c_recvExact.0:$(MAX_NETWORK_RECV_TRIES)
 # Unlike recvExact, sendPacket is not bounded by the timeout. The loop in
 # sendPacket will continue until all the bytes are sent or a network error
 # occurs. Please see NetworkInterfaceReceiveStub in

test/cbmc/proofs/MQTT_ReceiveLoop/Makefile

@@ -11,6 +11,10 @@ PROOF_UID=MQTT_ReceiveLoop
 # out of 2 we can get coverage of the entire function. Another iteration will
 # performed unnecessarily duplicating of the proof.
 MQTT_RECEIVE_TIMEOUT=3
+# The NetworkInterfaceReceiveStub is called once for getting the incoming packet
+# type with one byte of data, then it is called multiple times to reveive the
+# packet.
+MAX_NETWORK_RECV_TRIES=4
 # Please see test/cbmc/stubs/network_interface_subs.c for
 # more information on MAX_NETWORK_SEND_TRIES.
 MAX_NETWORK_SEND_TRIES=3
@@ -19,6 +23,7 @@ MAX_NETWORK_SEND_TRIES=3
 MQTT_STATE_ARRAY_MAX_COUNT=11
 DEFINES += -DMQTT_RECEIVE_TIMEOUT=$(MQTT_RECEIVE_TIMEOUT)
 DEFINES += -DMAX_NETWORK_SEND_TRIES=$(MAX_NETWORK_SEND_TRIES)
+DEFINES += -DMAX_NETWORK_RECV_TRIES=$(MAX_NETWORK_RECV_TRIES)
 INCLUDES +=
 
 # These functions have their memory saftey proven in other harnesses.
@@ -28,7 +33,7 @@ REMOVE_FUNCTION_BODY += MQTT_SerializeAck
 # The loops below are unwound once more than the exclusive timeout bound.
 UNWINDSET += MQTT_ReceiveLoop.0:$(MQTT_RECEIVE_TIMEOUT)
 UNWINDSET += __CPROVER_file_local_core_mqtt_c_discardPacket.0:$(MQTT_RECEIVE_TIMEOUT)
-UNWINDSET += __CPROVER_file_local_core_mqtt_c_recvExact.0:$(MQTT_RECEIVE_TIMEOUT)
+UNWINDSET += __CPROVER_file_local_core_mqtt_c_recvExact.0:$(MAX_NETWORK_RECV_TRIES)
 # Unlike recvExact, sendPacket is not bounded by the timeout. The loop in
 # sendPacket will continue until all the bytes are sent or a network error
 # occurs. Please see NetworkInterfaceReceiveStub in

test/cbmc/stubs/network_interface_stubs.c

@@ -35,6 +35,13 @@
     #define MAX_NETWORK_SEND_TRIES    3
 #endif
 
+/* An exclusive bound on the times that the NetworkInterfaceReceiveStub will
+ * return an unbound value. At this value and beyond, the
+ * NetworkInterfaceReceiveStub will return zero on every call. */
+#ifndef MAX_NETWORK_RECV_TRIES
+    #define MAX_NETWORK_RECV_TRIES    4
+#endif
+
 int32_t NetworkInterfaceReceiveStub( NetworkContext_t * pNetworkContext,
                                      void * pBuffer,
                                      size_t bytesToRecv )
@@ -48,11 +55,21 @@ int32_t NetworkInterfaceReceiveStub( NetworkContext_t * pNetworkContext,
     __CPROVER_havoc_object( pBuffer );
 
     int32_t bytesOrError;
+    static size_t tries = 0;
 
     /* It is a bug for the application defined transport send function to return
      * more than bytesToRecv. */
     __CPROVER_assume( bytesOrError <= ( int32_t ) bytesToRecv );
 
+    if( tries < ( MAX_NETWORK_RECV_TRIES - 1 ) )
+    {
+        tries++;
+    }
+    else
+    {
+        bytesOrError = 0;
+    }
+
     return bytesOrError;
 }