You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/10/09 13:05:43 UTC
(iggy) 01/01: fix(go): reject malformed user headers in NewIggyMessage
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch go_header_val
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit b2e3763fd09880228013a653c6f9f6d45d37cf57
Author: Grzegorz Koszyk <g....@gmail.com>
AuthorDate: Fri Oct 9 15:05:31 2026 +0200
fix(go): reject malformed user headers in NewIggyMessage
---
bdd/go/tests/tcp_test/messages_steps.go | 6 +-
foreign/go/contracts/messages.go | 7 ++
foreign/go/contracts/messages_test.go | 151 ++++++++++++++++++++++++++++
foreign/go/contracts/user_headers.go | 88 ++++++++++++++--
foreign/go/internal/command/message_test.go | 10 +-
foreign/go/internal/command/topic_test.go | 35 +++++++
6 files changed, 286 insertions(+), 11 deletions(-)
diff --git a/bdd/go/tests/tcp_test/messages_steps.go b/bdd/go/tests/tcp_test/messages_steps.go
index 6908b1b14..4be24ba70 100644
--- a/bdd/go/tests/tcp_test/messages_steps.go
+++ b/bdd/go/tests/tcp_test/messages_steps.go
@@ -37,8 +37,10 @@ func createDefaultMessageHeaders() []iggcon.HeaderEntry {
func createDefaultMessages() []iggcon.IggyMessage {
headers := createDefaultMessageHeaders()
- msg1, _ := iggcon.NewIggyMessage([]byte(createRandomString(256)), iggcon.WithID(uuid.New()), iggcon.WithUserHeaders(headers))
- msg2, _ := iggcon.NewIggyMessage([]byte(createRandomString(256)), iggcon.WithID(uuid.New()), iggcon.WithUserHeaders(headers))
+ msg1, err := iggcon.NewIggyMessage([]byte(createRandomString(256)), iggcon.WithID(uuid.New()), iggcon.WithUserHeaders(headers))
+ gomega.Expect(err).To(gomega.BeNil())
+ msg2, err := iggcon.NewIggyMessage([]byte(createRandomString(256)), iggcon.WithID(uuid.New()), iggcon.WithUserHeaders(headers))
+ gomega.Expect(err).To(gomega.BeNil())
return []iggcon.IggyMessage{msg1, msg2}
}
diff --git a/foreign/go/contracts/messages.go b/foreign/go/contracts/messages.go
index faa64508e..60dbdb24b 100644
--- a/foreign/go/contracts/messages.go
+++ b/foreign/go/contracts/messages.go
@@ -42,6 +42,7 @@ const (
// - Maximum headers size: 100 KB
// - Each individual header key is limited to 255 bytes
// - Each individual header value is limited to 255 bytes
+ // - Keys and values must not be empty, and a fixed-size kind must have its exact width
MaxUserHeadersSize = 100 * 1000
)
@@ -84,6 +85,9 @@ type IggyMessage struct {
type IggyMessageOpt func(message *IggyMessage)
// NewIggyMessage Creates a new message with customizable parameters.
+//
+// It returns ierror.ErrInvalidHeaderKey or ierror.ErrInvalidHeaderValue for a user header that readers would refuse
+// to decode, since readers drop or fail on every header of a message that carries one.
func NewIggyMessage(payload []byte, opts ...IggyMessageOpt) (IggyMessage, error) {
if len(payload) == 0 {
return IggyMessage{}, ierror.ErrInvalidMessagePayloadLength
@@ -108,6 +112,9 @@ func NewIggyMessage(payload []byte, opts ...IggyMessageOpt) (IggyMessage, error)
if userHeaderLength > MaxUserHeadersSize {
return IggyMessage{}, ierror.ErrTooBigUserHeaders
}
+ if err := validateUserHeaders(message.UserHeaders); err != nil {
+ return IggyMessage{}, err
+ }
message.Header.UserHeaderLength = uint32(userHeaderLength)
return message, nil
}
diff --git a/foreign/go/contracts/messages_test.go b/foreign/go/contracts/messages_test.go
new file mode 100644
index 000000000..982a3ffd3
--- /dev/null
+++ b/foreign/go/contracts/messages_test.go
@@ -0,0 +1,151 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package iggcon
+
+import (
+ "bytes"
+ "errors"
+ "strings"
+ "testing"
+
+ ierror "github.com/apache/iggy/foreign/go/errors"
+)
+
+// The user-header wire format bounds every key and value to 1..=255 bytes and
+// a fixed-size kind to its exact width. Every reader enforces that: this
+// package's DeserializeHeaders, and the Rust SDK and server, which drop the
+// whole header map of a message that breaks it. The server stores header bytes
+// as sent (they may be client-side encrypted, so it cannot check them), so a
+// message NewIggyMessage accepts must already be one every reader can decode.
+func TestNewIggyMessage_HeadersItAcceptsAreDecodable(t *testing.T) {
+ key := HeaderKey{Kind: String, Value: []byte("dedup-key")}
+ tests := []struct {
+ name string
+ entry HeaderEntry
+ valid bool
+ }{
+ {"1-byte string value", HeaderEntry{key, HeaderValue{Kind: String, Value: []byte("a")}}, true},
+ {"255-byte raw value", HeaderEntry{key, HeaderValue{Kind: Raw, Value: bytes.Repeat([]byte{7}, 255)}}, true},
+ {"4-byte int32 value", HeaderEntry{key, HeaderValue{Kind: Int32, Value: []byte{1, 0, 0, 0}}}, true},
+ {"8-byte double value", HeaderEntry{key, HeaderValue{Kind: Double, Value: make([]byte, 8)}}, true},
+ {"empty string value", HeaderEntry{key, HeaderValue{Kind: String, Value: []byte{}}}, false},
+ {"nil raw value", HeaderEntry{key, HeaderValue{Kind: Raw}}, false},
+ {"256-byte raw value", HeaderEntry{key, HeaderValue{Kind: Raw, Value: bytes.Repeat([]byte{7}, 256)}}, false},
+ {"3-byte int32 value", HeaderEntry{key, HeaderValue{Kind: Int32, Value: []byte{1, 0, 0}}}, false},
+ {"empty key", HeaderEntry{HeaderKey{Kind: String}, HeaderValue{Kind: String, Value: []byte("a")}}, false},
+ {"256-byte key", HeaderEntry{HeaderKey{Kind: Raw, Value: bytes.Repeat([]byte{7}, 256)}, HeaderValue{Kind: String, Value: []byte("a")}}, false},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ message, err := NewIggyMessage([]byte("payload"), WithUserHeaders([]HeaderEntry{tt.entry}))
+ if tt.valid {
+ if err != nil {
+ t.Fatalf("NewIggyMessage refused a well-formed header: %v", err)
+ }
+ decoded, err := DeserializeHeaders(message.UserHeaders)
+ if err != nil || len(decoded) != 1 || !bytes.Equal(decoded[0].Value.Value, tt.entry.Value.Value) {
+ t.Fatalf("header did not round-trip: %v %+v", err, decoded)
+ }
+ return
+ }
+ if err == nil {
+ _, decodeErr := DeserializeHeaders(message.UserHeaders)
+ t.Fatalf("NewIggyMessage accepted a header no reader can decode (DeserializeHeaders: %v)", decodeErr)
+ }
+ if !errors.Is(err, ierror.ErrInvalidHeaderValue) && !errors.Is(err, ierror.ErrInvalidHeaderKey) {
+ t.Fatalf("want InvalidHeaderKey/InvalidHeaderValue, got %v", err)
+ }
+ })
+ }
+}
+
+func TestNewIggyMessage_ReportsWhichHeaderFieldIsInvalid(t *testing.T) {
+ key := HeaderKey{Kind: String, Value: []byte("dedup-key")}
+ value := HeaderValue{Kind: String, Value: []byte("a")}
+ tests := []struct {
+ name string
+ entry HeaderEntry
+ want error
+ }{
+ {"empty value", HeaderEntry{key, HeaderValue{Kind: String}}, ierror.ErrInvalidHeaderValue},
+ {"3-byte int32 value", HeaderEntry{key, HeaderValue{Kind: Int32, Value: []byte{1, 0, 0}}}, ierror.ErrInvalidHeaderValue},
+ {"zero value kind", HeaderEntry{key, HeaderValue{Value: []byte("a")}}, ierror.ErrInvalidHeaderValue},
+ {"empty key", HeaderEntry{HeaderKey{Kind: String}, value}, ierror.ErrInvalidHeaderKey},
+ {"2-byte uint32 key", HeaderEntry{HeaderKey{Kind: Uint32, Value: []byte{1, 0}}, value}, ierror.ErrInvalidHeaderKey},
+ {"unknown key kind", HeaderEntry{HeaderKey{Kind: 16, Value: []byte("k")}, value}, ierror.ErrInvalidHeaderKey},
+ {"kind 257 key", HeaderEntry{HeaderKey{Kind: 257, Value: []byte("k")}, value}, ierror.ErrInvalidHeaderKey},
+ {"kind -1 value", HeaderEntry{key, HeaderValue{Kind: -1, Value: []byte("a")}}, ierror.ErrInvalidHeaderValue},
+ {"kind -255 value", HeaderEntry{key, HeaderValue{Kind: -255, Value: []byte("a")}}, ierror.ErrInvalidHeaderValue},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ _, err := NewIggyMessage([]byte("payload"), WithUserHeaders([]HeaderEntry{tt.entry}))
+ if !errors.Is(err, tt.want) {
+ t.Fatalf("want %v, got %v", tt.want, err)
+ }
+ })
+ }
+}
+
+func TestNewIggyMessage_RejectsMalformedRawHeaderBytes(t *testing.T) {
+ tests := []struct {
+ name string
+ raw []byte
+ want error
+ }{
+ {"short TLV", []byte{2, 1, 0}, ierror.ErrInvalidHeaderKey},
+ {"4-byte tail", []byte{2, 1, 0, 0}, ierror.ErrInvalidHeaderKey},
+ {"truncated data", []byte{2, 5, 0, 0, 0, 'k'}, ierror.ErrInvalidHeaderKey},
+ {"key with no value", []byte{2, 1, 0, 0, 0, 'k'}, ierror.ErrInvalidHeaderValue},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ _, err := NewIggyMessage([]byte("payload"), func(m *IggyMessage) { m.UserHeaders = tt.raw })
+ if !errors.Is(err, tt.want) {
+ t.Fatalf("want %v, got %v", tt.want, err)
+ }
+ })
+ }
+}
+
+func TestNewIggyMessage_NamesTheInvalidHeaderEntry(t *testing.T) {
+ good := HeaderEntry{
+ Key: HeaderKey{Kind: String, Value: []byte("k")},
+ Value: HeaderValue{Kind: String, Value: []byte("v")},
+ }
+ bad := HeaderEntry{Key: good.Key, Value: HeaderValue{Kind: String}}
+ _, err := NewIggyMessage([]byte("payload"), WithUserHeaders([]HeaderEntry{good, good, bad}))
+ if !errors.Is(err, ierror.ErrInvalidHeaderValue) {
+ t.Fatalf("want %v, got %v", ierror.ErrInvalidHeaderValue, err)
+ }
+ if !strings.Contains(err.Error(), "header entry 2") {
+ t.Fatalf("error %q does not name entry 2", err)
+ }
+ if !strings.Contains(err.Error(), headerFieldLengthOutOfRange) {
+ t.Fatalf("error %q does not give the reason %q", err, headerFieldLengthOutOfRange)
+ }
+}
+
+func TestNewHeaderKey_RejectsOutOfRangeLengthAsInvalidHeaderKey(t *testing.T) {
+ if _, err := NewHeaderKeyString(""); !errors.Is(err, ierror.ErrInvalidHeaderKey) {
+ t.Errorf("NewHeaderKeyString(\"\"): want %v, got %v", ierror.ErrInvalidHeaderKey, err)
+ }
+ if _, err := NewHeaderKeyRaw(make([]byte, maxHeaderFieldLength+1)); !errors.Is(err, ierror.ErrInvalidHeaderKey) {
+ t.Errorf("NewHeaderKeyRaw(256 bytes): want %v, got %v", ierror.ErrInvalidHeaderKey, err)
+ }
+}
diff --git a/foreign/go/contracts/user_headers.go b/foreign/go/contracts/user_headers.go
index 2d678063f..33e270d2d 100644
--- a/foreign/go/contracts/user_headers.go
+++ b/foreign/go/contracts/user_headers.go
@@ -20,8 +20,13 @@ package iggcon
import (
"encoding/binary"
"errors"
+ "fmt"
+
+ ierror "github.com/apache/iggy/foreign/go/errors"
)
+const maxHeaderFieldLength = 255
+
type HeaderValue struct {
Kind HeaderKind
Value []byte
@@ -38,15 +43,15 @@ type HeaderEntry struct {
}
func NewHeaderKeyString(val string) (HeaderKey, error) {
- if len(val) == 0 || len(val) > 255 {
- return HeaderKey{}, errors.New("value has incorrect size, must be between 1 and 255")
+ if len(val) == 0 || len(val) > maxHeaderFieldLength {
+ return HeaderKey{}, fmt.Errorf("%w: length must be between 1 and %d", ierror.ErrInvalidHeaderKey, maxHeaderFieldLength)
}
return HeaderKey{Kind: String, Value: []byte(val)}, nil
}
func NewHeaderKeyRaw(val []byte) (HeaderKey, error) {
- if len(val) == 0 || len(val) > 255 {
- return HeaderKey{}, errors.New("value has incorrect size, must be between 1 and 255")
+ if len(val) == 0 || len(val) > maxHeaderFieldLength {
+ return HeaderKey{}, fmt.Errorf("%w: length must be between 1 and %d", ierror.ErrInvalidHeaderKey, maxHeaderFieldLength)
}
return HeaderKey{Kind: Raw, Value: val}, nil
}
@@ -94,6 +99,20 @@ func (k HeaderKind) ExpectedSize() int {
}
}
+// The defined range must move together with WireHeaderKind::KNOWN_MAX in
+// core/binary_protocol/src/primitives/user_headers.rs.
+func (k HeaderKind) valid() bool {
+ return k >= Raw && k <= Double
+}
+
+// Out-of-byte kinds would alias a defined kind once truncated, so they encode as the reserved 0 that the Rust reader and the server reject.
+func (k HeaderKind) wireByte() byte {
+ if k < 0 || k > 255 {
+ return 0
+ }
+ return byte(k)
+}
+
func GetHeadersBytes(headers []HeaderEntry) []byte {
headersLength := 0
for _, entry := range headers {
@@ -114,7 +133,7 @@ func getBytesFromHeader(key HeaderKey, value HeaderValue) []byte {
headerBytes := make([]byte, headerBytesLength)
pos := 0
- headerBytes[pos] = byte(key.Kind)
+ headerBytes[pos] = key.Kind.wireByte()
pos++
binary.LittleEndian.PutUint32(headerBytes[pos:pos+4], uint32(len(key.Value)))
@@ -122,7 +141,7 @@ func getBytesFromHeader(key HeaderKey, value HeaderValue) []byte {
copy(headerBytes[pos:pos+len(key.Value)], key.Value)
pos += len(key.Value)
- headerBytes[pos] = byte(value.Kind)
+ headerBytes[pos] = value.Kind.wireByte()
pos++
binary.LittleEndian.PutUint32(headerBytes[pos:pos+4], uint32(len(value.Value)))
@@ -203,6 +222,63 @@ func DeserializeHeaders(userHeadersBytes []byte) ([]HeaderEntry, error) {
return headers, nil
}
+const (
+ headerFieldTruncated = "truncated"
+ headerFieldUnknownKind = "unknown kind"
+ headerFieldLengthOutOfRange = "length out of range"
+ headerFieldWrongWidth = "wrong width for kind"
+)
+
+// validateUserHeaders rejects any field some reader refuses to decode.
+func validateUserHeaders(userHeadersBytes []byte) error {
+ position := 0
+ entry := 0
+ isKey := true
+ for position < len(userHeadersBytes) {
+ next, reason := headerFieldEnd(userHeadersBytes, position)
+ if reason != "" {
+ if isKey {
+ return fmt.Errorf("header entry %d: %w: %s", entry, ierror.ErrInvalidHeaderKey, reason)
+ }
+ return fmt.Errorf("header entry %d: %w: %s", entry, ierror.ErrInvalidHeaderValue, reason)
+ }
+ position = next
+ if !isKey {
+ entry++
+ }
+ isKey = !isKey
+ }
+ if !isKey {
+ return fmt.Errorf("header entry %d: %w: %s", entry, ierror.ErrInvalidHeaderValue, headerFieldTruncated)
+ }
+ return nil
+}
+
+// headerFieldEnd returns the position after the kind-length-data field at position, or the reason a reader would reject it.
+func headerFieldEnd(userHeadersBytes []byte, position int) (int, string) {
+ if len(userHeadersBytes)-position < 5 {
+ return 0, headerFieldTruncated
+ }
+ field := userHeadersBytes[position : position+5]
+ kind := HeaderKind(field[0])
+ fieldLength := binary.LittleEndian.Uint32(field[1:5])
+ position += 5
+ if !kind.valid() {
+ return 0, headerFieldUnknownKind
+ }
+ if fieldLength == 0 || fieldLength > maxHeaderFieldLength {
+ return 0, headerFieldLengthOutOfRange
+ }
+ length := int(fieldLength)
+ if len(userHeadersBytes)-position < length {
+ return 0, headerFieldTruncated
+ }
+ if expected := kind.ExpectedSize(); expected != -1 && length != expected {
+ return 0, headerFieldWrongWidth
+ }
+ return position + length, ""
+}
+
func deserializeHeaderKind(payload []byte, position int) (HeaderKind, error) {
if position >= len(payload) {
return 0, errors.New("invalid header kind position")
diff --git a/foreign/go/internal/command/message_test.go b/foreign/go/internal/command/message_test.go
index d6944d6a7..5554932e0 100644
--- a/foreign/go/internal/command/message_test.go
+++ b/foreign/go/internal/command/message_test.go
@@ -142,7 +142,7 @@ func TestSerialize_SendMessagesMatchesTheGoldenVector(t *testing.T) {
}
func TestSerialize_SendMessagesRequest(t *testing.T) {
- message1 := generateTestMessage("data1")
+ message1 := generateTestMessage(t, "data1")
streamId, _ := iggcon.NewIdentifier("test_stream_id")
topicId, _ := iggcon.NewIdentifier("test_topic_id")
request := SendMessages{
@@ -370,10 +370,14 @@ func createDefaultMessageHeaders() []iggcon.HeaderEntry {
}
}
-func generateTestMessage(payload string) iggcon.IggyMessage {
- msg, _ := iggcon.NewIggyMessage(
+func generateTestMessage(t *testing.T, payload string) iggcon.IggyMessage {
+ t.Helper()
+ msg, err := iggcon.NewIggyMessage(
[]byte(payload),
iggcon.WithID(uuid.New()),
iggcon.WithUserHeaders(createDefaultMessageHeaders()))
+ if err != nil {
+ t.Fatalf("failed to create message: %v", err)
+ }
return msg
}
diff --git a/foreign/go/internal/command/topic_test.go b/foreign/go/internal/command/topic_test.go
index 2c450a044..4583c0262 100644
--- a/foreign/go/internal/command/topic_test.go
+++ b/foreign/go/internal/command/topic_test.go
@@ -284,3 +284,38 @@ func TestCreateTopic_NonStringOptionKeyIsRejected(t *testing.T) {
t.Error("expected an error for a non-string option key, got nil")
}
}
+
+func TestCreateTopic_OutOfByteOptionValueKindEncodesAsReservedZero(t *testing.T) {
+ streamId, err := iggcon.NewIdentifier(uint32(1))
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ command := CreateTopic{
+ StreamId: streamId,
+ Name: "t",
+ PartitionsCount: 1,
+ Options: []iggcon.HeaderEntry{{
+ Key: iggcon.HeaderKey{Kind: iggcon.String, Value: []byte("future_option")},
+ Value: iggcon.HeaderValue{Kind: 257, Value: []byte("x")},
+ }},
+ }
+
+ serialized, err := command.MarshalBinary()
+ if err != nil {
+ t.Fatalf("failed to serialize CreateTopic: %v", err)
+ }
+
+ const streamIdSize = 2 + 4
+ position := streamIdSize + 4 + 1 + len(command.Name)
+ block := serialized[position:]
+ key := command.Options[0].Key.Value
+ entry := append(append([]byte{byte(iggcon.String)}, binary.LittleEndian.AppendUint32(nil, uint32(len(key)))...), key...)
+ index := bytes.Index(block, entry)
+ if index < 0 {
+ t.Fatal("the caller's option entry is missing from the options block")
+ }
+ // 257 truncates to 1, which would let a reader decode the value as Raw.
+ if kind := block[index+len(entry)]; kind != 0 {
+ t.Fatalf("value kind byte = %d, want the reserved 0", kind)
+ }
+}