[C] Add aeron_is_frame_valid method.

Author: vyazelenko

aeron-client/src/main/c/protocol/aeron_udp_protocol.c

@@ -43,6 +43,8 @@ int aeron_udp_protocol_group_tag(aeron_status_message_header_t *sm, int64_t *gro
     return (int)((size_t)sm->frame_header.frame_length - group_tag_offset);
 }
 
+extern bool aeron_is_frame_valid(const aeron_frame_header_t *header, size_t frame_length);
+
 extern size_t aeron_res_header_address_length(int8_t res_type);
 
 extern size_t aeron_compute_max_message_length(size_t term_length);

aeron-client/src/main/c/protocol/aeron_udp_protocol.h

@@ -17,6 +17,7 @@
 #ifndef AERON_UDP_PROTOCOL_H
 #define AERON_UDP_PROTOCOL_H
 
+#include <stdbool.h>
 #include <stdint.h>
 #include <stddef.h>
 
@@ -233,6 +234,11 @@ int aeron_udp_protocol_group_tag(aeron_status_message_header_t *sm, int64_t *gro
 #define AERON_ERROR_MAX_FRAME_LENGTH (AERON_ERROR_HEADER_LENGTH + AERON_ERROR_MAX_TEXT_LENGTH)
 #define AERON_ERROR_HAS_GROUP_TAG_FLAG (0x08)
 
+inline bool aeron_is_frame_valid(const aeron_frame_header_t *header, const size_t frame_length)
+{
+    return frame_length >= AERON_FRAME_HEADER_LENGTH && AERON_FRAME_HEADER_VERSION == header->version;
+}
+
 inline size_t aeron_res_header_address_length(int8_t res_type)
 {
     return AERON_RES_HEADER_TYPE_NAME_TO_IP6_MD == res_type ?
@@ -241,8 +247,7 @@ inline size_t aeron_res_header_address_length(int8_t res_type)
 
 inline size_t aeron_compute_max_message_length(size_t term_length)
 {
-    size_t max_length_for_term = term_length / 8;
-
+    size_t max_length_for_term = term_length >> 3;
     return max_length_for_term < AERON_FRAME_MAX_MESSAGE_LENGTH ? max_length_for_term : AERON_FRAME_MAX_MESSAGE_LENGTH;
 }
 

aeron-driver/src/main/c/aeron_driver_name_resolver.c

@@ -602,7 +602,7 @@ void aeron_driver_name_resolver_receive(
     aeron_frame_header_t *frame_header = (aeron_frame_header_t *)buffer;
     size_t remaining = length;
 
-    if ((remaining < AERON_FRAME_HEADER_LENGTH) || (frame_header->version != AERON_FRAME_HEADER_VERSION))
+    if (!aeron_is_frame_valid(frame_header, length) || length < (size_t)frame_header->frame_length)
     {
         aeron_counter_increment(resolver->invalid_packets_counter);
         return;

aeron-driver/src/main/c/media/aeron_receive_channel_endpoint.c

@@ -487,7 +487,7 @@ void aeron_receive_channel_endpoint_dispatch(
     aeron_receive_channel_endpoint_t *endpoint = (aeron_receive_channel_endpoint_t *)endpoint_clientd;
     aeron_receive_destination_t *destination = (aeron_receive_destination_t *)destination_clientd;
 
-    if ((length < AERON_FRAME_HEADER_LENGTH) || (frame_header->version != AERON_FRAME_HEADER_VERSION))
+    if (!aeron_is_frame_valid(frame_header, length))
     {
         aeron_counter_increment(receiver->invalid_frames_counter);
         return;
@@ -513,7 +513,7 @@ void aeron_receive_channel_endpoint_dispatch(
             break;
 
         case AERON_HDR_TYPE_SETUP:
-            if (length >= AERON_SETUP_HEADER_LENGTH)
+            if (length >= AERON_SETUP_HEADER_LENGTH && length >= (size_t)frame_header->frame_length)
             {
                 if (aeron_receive_channel_endpoint_on_setup(endpoint, destination, buffer, length, addr) < 0)
                 {
@@ -528,7 +528,7 @@ void aeron_receive_channel_endpoint_dispatch(
             break;
 
         case AERON_HDR_TYPE_RTTM:
-            if (length >= AERON_RTTM_HEADER_LENGTH)
+            if (length >= AERON_RTTM_HEADER_LENGTH && length >= (size_t)frame_header->frame_length)
             {
                 if (aeron_receive_channel_endpoint_on_rttm(endpoint, destination, buffer, length, addr) < 0)
                 {

aeron-driver/src/main/c/media/aeron_send_channel_endpoint.c

@@ -419,10 +419,16 @@ void aeron_send_channel_endpoint_dispatch(
 
     int result = 0;
 
+    if (!aeron_is_frame_valid(frame_header, length))
+    {
+        aeron_counter_increment(sender->invalid_frames_counter);
+        return;
+    }
+
     switch (frame_header->type)
     {
         case AERON_HDR_TYPE_NAK:
-            if (length >= AERON_NAK_HEADER_LENGTH)
+            if (length >= AERON_NAK_HEADER_LENGTH && length >= (size_t)frame_header->frame_length)
             {
                 result = aeron_send_channel_endpoint_on_nak(endpoint, buffer, length, addr);
                 aeron_counter_increment_release(sender->nak_messages_received_counter);
@@ -458,7 +464,7 @@ void aeron_send_channel_endpoint_dispatch(
             break;
 
         case AERON_HDR_TYPE_RTTM:
-            if (length >= AERON_RTTM_HEADER_LENGTH)
+            if (length >= AERON_RTTM_HEADER_LENGTH && length >= (size_t)frame_header->frame_length)
             {
                 aeron_send_channel_endpoint_on_rttm(endpoint, buffer, length, addr);
             }
@@ -469,7 +475,7 @@ void aeron_send_channel_endpoint_dispatch(
             break;
 
         case AERON_HDR_TYPE_RSP_SETUP:
-            if (length >= AERON_RSP_SETUP_HEADER_LENGTH)
+            if (length >= AERON_RSP_SETUP_HEADER_LENGTH && length >= (size_t)frame_header->frame_length)
             {
                 aeron_send_channel_endpoint_on_response_setup(endpoint, conductor_proxy, buffer, length, addr);
             }

aeron-driver/src/test/c/aeron_udp_channel_test.cpp

@@ -136,6 +136,34 @@ TEST_F(UdpChannelTest, constantSanityChecks)
     EXPECT_EQ(AERON_RSP_SETUP_HEADER_LENGTH, sizeof(aeron_response_setup_header_t));
 }
 
+TEST_F(UdpChannelTest, shouldComputeMaxMessageLength)
+{
+    EXPECT_EQ(0, aeron_compute_max_message_length(0));
+    EXPECT_EQ(4, aeron_compute_max_message_length(32));
+    EXPECT_EQ(8192, aeron_compute_max_message_length(64 * 1024));
+    EXPECT_EQ(8 * 1024 * 1024, aeron_compute_max_message_length(64 * 1024 * 1024));
+    EXPECT_EQ(AERON_FRAME_MAX_MESSAGE_LENGTH, aeron_compute_max_message_length(1024 * 1024 * 1024));
+    EXPECT_EQ(16 * 1024 * 1024, aeron_compute_max_message_length(512 * 1024 * 1024));
+}
+
+TEST_F(UdpChannelTest, shouldCheckIfFrameIsValid)
+{
+    aeron_frame_header_t header = {};
+
+    EXPECT_FALSE(aeron_is_frame_valid(&header, 1)); // length below min header length
+
+    header.version = 123;
+    EXPECT_FALSE(aeron_is_frame_valid(&header, AERON_FRAME_HEADER_LENGTH)); // wrong version
+
+    header.version = AERON_FRAME_HEADER_VERSION;
+    header.frame_length = 100;
+    EXPECT_TRUE(aeron_is_frame_valid(&header, AERON_FRAME_HEADER_LENGTH + 1));
+    header.type = 123;
+    EXPECT_TRUE(aeron_is_frame_valid(&header, header.frame_length));
+    header.flags = 0xAE;
+    EXPECT_TRUE(aeron_is_frame_valid(&header, INT32_MAX));
+}
+
 TEST_F(UdpChannelTest, shouldParseExplicitLocalAddressAndPortFormat)
 {
     ASSERT_EQ(parse_udp_channel("aeron:udp?interface=localhost:40123|endpoint=localhost:40124"), 0) << aeron_errmsg();