GH-55400: [C++][IPC] Validate body_length in ReadMessage and ReadMessageAsync

cpp/src/arrow/ipc/message.cc

@@ -37,6 +37,7 @@
 #include "arrow/status.h"
 #include "arrow/util/endian.h"
 #include "arrow/util/future.h"
+#include "arrow/util/int_util_overflow.h"
 #include "arrow/util/logging_internal.h"
 #include "arrow/util/ubsan.h"
 
@@ -461,11 +462,22 @@ static Result<std::unique_ptr<Message>> ReadMessageInternal(
     return Status::Invalid("metadata_length should be at least ",
                            state.decoder->next_required_size());
   }
+
+  int64_t total_length = metadata_length;
+  if (body_length.has_value()) {
+    if (body_length.value() < 0) {
+      return Status::Invalid("body_length should be non-negative");
+    }
+    if (internal::AddWithOverflow(static_cast<int64_t>(metadata_length),
+                                  body_length.value(), &total_length)) {
+      return Status::Invalid("body_length overflows when added to metadata_length");
+    }
+  }
+
   // When body_length is known, read metadata + body in one IO call.
   // Otherwise, read only metadata first.
   ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Buffer> metadata,
-                        file->ReadAt(offset, metadata_length + body_length.value_or(0),
-                                     /*allow_short_read=*/false));
+                        file->ReadAt(offset, total_length, /*allow_short_read=*/false));
 
   return ReadMessageContinued(offset, metadata_length, body_length, metadata, file,
                               fields_loader, &state);
@@ -498,9 +510,17 @@ Future<std::shared_ptr<Message>> ReadMessageAsync(int64_t offset, int32_t metada
     return Status::Invalid("metadata_length should be at least ",
                            state->decoder->next_required_size());
   }
+  if (body_length < 0) {
+    return Status::Invalid("body_length should be non-negative");
+  }
+  int64_t total_length = 0;
+  if (internal::AddWithOverflow(static_cast<int64_t>(metadata_length), body_length,
+                                &total_length)) {
+    return Status::Invalid("body_length overflows when added to metadata_length");
+  }
+
   return file
-      ->ReadAsync(context, offset, metadata_length + body_length,
-                  /*allow_short_read=*/false)
+      ->ReadAsync(context, offset, total_length, /*allow_short_read=*/false)
       .Then([=](std::shared_ptr<Buffer> metadata) -> Result<std::shared_ptr<Message>> {
         // Pass a nullptr file to ensure that no further IO occurs
         // (we have fetched all the required bytes).

cpp/src/arrow/ipc/read_write_test.cc

@@ -640,6 +640,42 @@ TEST(TestReadMessage, ReadBodyWithLength) {
   ASSERT_TRUE(message->Verify());
 }
 
+TEST(TestReadMessage, ReadBodyWithInvalidLength) {
+  std::shared_ptr<RecordBatch> batch;
+  ASSERT_OK(MakeIntRecordBatch(&batch));
+
+  ASSERT_OK_AND_ASSIGN(auto stream, io::BufferOutputStream::Create(0));
+  int32_t metadata_length;
+  int64_t body_length;
+  ASSERT_OK(WriteRecordBatch(*batch, 0, stream.get(), &metadata_length, &body_length,
+                             IpcWriteOptions::Defaults()));
+
+  ASSERT_OK_AND_ASSIGN(auto buffer, stream->Finish());
+  io::BufferReader reader(buffer);
+
+  // Negative body_length should fail
+  ASSERT_RAISES_WITH_MESSAGE_THAT(
+      Invalid, ::testing::HasSubstr("body_length should be non-negative"),
+      ReadMessage(0, metadata_length, -1, &reader));
+
+  // Overflowing body_length should fail
+  ASSERT_RAISES_WITH_MESSAGE_THAT(
+      Invalid,
+      ::testing::HasSubstr("body_length overflows when added to metadata_length"),
+      ReadMessage(0, metadata_length, std::numeric_limits<int64_t>::max(), &reader));
+
+  // Async overload should also validate body_length
+  ASSERT_RAISES_WITH_MESSAGE_THAT(
+      Invalid, ::testing::HasSubstr("body_length should be non-negative"),
+      ReadMessageAsync(0, metadata_length, -1, &reader).status());
+
+  ASSERT_RAISES_WITH_MESSAGE_THAT(
+      Invalid,
+      ::testing::HasSubstr("body_length overflows when added to metadata_length"),
+      ReadMessageAsync(0, metadata_length, std::numeric_limits<int64_t>::max(), &reader)
+          .status());
+}
+
 TEST(TestMetadata, GetMetadataVersion) {
   ASSERT_EQ(MetadataVersion::V1, ipc::internal::GetMetadataVersion(
                                      flatbuf::MetadataVersion::MetadataVersion_V1));