This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-go.git
The following commit(s) were added to refs/heads/main by this push:
new 16ac2bc3 fix(ipc): reject truncated message frames (#991)
16ac2bc3 is described below
commit 16ac2bc3bb0ec332b482c5327db874b2442d0694
Author: Minh Vu <[email protected]>
AuthorDate: Thu Jul 23 18:03:35 2026 +0200
fix(ipc): reject truncated message frames (#991)
### Rationale for this change
Once an IPC message has started, a missing length, metadata block, or
body is a truncated frame rather than a clean end of stream.
`io.ReadFull` returns an error matching `io.EOF` when none of the
requested bytes are available, and the IPC stream reader treats errors
matching `io.EOF` as normal completion. As a result, some truncated
messages were silently accepted.
### What changes are included in this PR?
Normalize errors matching `io.EOF` to `io.ErrUnexpectedEOF` when a
required message component is missing after framing has begun. EOF
before the next continuation indicator and explicit EOS markers keep
their existing behavior.
Add coverage for truncation at the message length, metadata, and body,
including wrapped EOF errors, plus a stream-level regression test that
verifies the error reaches `Reader.Err()`.
### Are these changes tested?
Yes. `go test ./arrow/ipc` and focused race tests pass.
### Are there any user-facing changes?
Malformed IPC streams now report an unexpected EOF instead of appearing
to end successfully. Valid streams are unchanged.
---
arrow/ipc/message.go | 15 ++++++++++---
arrow/ipc/message_test.go | 56 +++++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 68 insertions(+), 3 deletions(-)
diff --git a/arrow/ipc/message.go b/arrow/ipc/message.go
index df1ed8d9..0f296db6 100644
--- a/arrow/ipc/message.go
+++ b/arrow/ipc/message.go
@@ -18,6 +18,7 @@ package ipc
import (
"encoding/binary"
+ "errors"
"fmt"
"io"
"sync/atomic"
@@ -172,6 +173,14 @@ func NewMessageReader(r io.Reader, opts ...Option)
MessageReader {
return mr
}
+func readFullMessagePart(r io.Reader, buf []byte) error {
+ _, err := io.ReadFull(r, buf)
+ if errors.Is(err, io.EOF) {
+ return io.ErrUnexpectedEOF
+ }
+ return err
+}
+
// Retain increases the reference count by 1.
// Retain may be called simultaneously from multiple goroutines.
func (r *messageReader) Retain() {
@@ -217,7 +226,7 @@ func (r *messageReader) Message() (msg *Message, err error)
{
// EOS message.
return nil, io.EOF // FIXME(sbinet): send nil instead? or a
special EOS error?
case kIPCContToken:
- _, err = io.ReadFull(r.r, buf)
+ err = readFullMessagePart(r.r, buf)
if err != nil {
return nil, fmt.Errorf("arrow/ipc: could not read
message length: %w", err)
}
@@ -240,7 +249,7 @@ func (r *messageReader) Message() (msg *Message, err error)
{
}
buf = make([]byte, msgLen)
- _, err = io.ReadFull(r.r, buf)
+ err = readFullMessagePart(r.r, buf)
if err != nil {
return nil, fmt.Errorf("arrow/ipc: could not read message
metadata: %w", err)
}
@@ -259,7 +268,7 @@ func (r *messageReader) Message() (msg *Message, err error)
{
defer body.Release()
body.Resize(int(bodyLen))
- _, err = io.ReadFull(r.r, body.Bytes())
+ err = readFullMessagePart(r.r, body.Bytes())
if err != nil {
return nil, fmt.Errorf("arrow/ipc: could not read message body:
%w", err)
}
diff --git a/arrow/ipc/message_test.go b/arrow/ipc/message_test.go
index 896a5bc4..0498379f 100644
--- a/arrow/ipc/message_test.go
+++ b/arrow/ipc/message_test.go
@@ -25,6 +25,7 @@ import (
"math"
"strconv"
"testing"
+ "testing/iotest"
"github.com/apache/arrow-go/v18/arrow"
"github.com/apache/arrow-go/v18/arrow/array"
@@ -102,6 +103,61 @@ func TestMessageReaderRejectsInvalidBodyLengths(t
*testing.T) {
}
}
+func TestMessageReaderRejectsTruncatedMessages(t *testing.T) {
+ continuationToken := make([]byte, 4)
+ binary.LittleEndian.PutUint32(continuationToken, uint32(kIPCContToken))
+
+ metadataLength := append([]byte(nil), continuationToken...)
+ metadataLength = binary.LittleEndian.AppendUint32(metadataLength, 4)
+
+ tests := []struct {
+ name string
+ input io.Reader
+ }{
+ {name: "message length", input:
bytes.NewBuffer(continuationToken)},
+ {
+ name: "wrapped EOF after continuation token",
+ input: io.MultiReader(
+ bytes.NewReader(continuationToken),
+ iotest.ErrReader(fmt.Errorf("wrapped: %w",
io.EOF)),
+ ),
+ },
+ {name: "message metadata", input:
bytes.NewBuffer(metadataLength)},
+ {name: "message body", input: messageReaderInput(t, 1)},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ r := NewMessageReader(tt.input)
+ defer r.Release()
+
+ _, err := r.Message()
+ require.ErrorIs(t, err, io.ErrUnexpectedEOF)
+ require.NotErrorIs(t, err, io.EOF)
+ })
+ }
+}
+
+func TestMessageReaderAllowsEndOfStreamAtMessageBoundary(t *testing.T) {
+ r := NewMessageReader(bytes.NewReader(nil))
+ defer r.Release()
+
+ _, err := r.Message()
+ require.ErrorIs(t, err, io.EOF)
+}
+
+func TestReaderReportsTruncatedMessage(t *testing.T) {
+ input := writeRecordsIntoBuffer(t, 0)
+ input.Truncate(input.Len() - 4)
+
+ r, err := NewReader(input)
+ require.NoError(t, err)
+ defer r.Release()
+
+ require.False(t, r.Next())
+ require.ErrorIs(t, r.Err(), io.ErrUnexpectedEOF)
+}
+
func TestBodySizeLimitConfig(t *testing.T) {
require.Equal(t, defaultMaxBodySize, newConfig().maxBodySize)
require.Zero(t, newConfig(WithBodySizeLimit(0)).maxBodySize)
