(iggy) branch master updated: fix(go): check payload length in DeserializeOffset (#2135)

[hg...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

hgruszecki pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git


The following commit(s) were added to refs/heads/master by this push:
     new d2ac5480 fix(go): check payload length in DeserializeOffset (#2135)
d2ac5480 is described below

commit d2ac5480b5641d13754c737f2563110b50f8760a
Author: haze518 <42...@users.noreply.github.com>
AuthorDate: Sat Aug 30 18:06:31 2025 +0600

    fix(go): check payload length in DeserializeOffset (#2135)
    
    Added integration tests for consumer offset scenarios and payload
    validation in `DeserializeOffset` to prevent slice bounds panics with
    empty payloads.
    
    Changes:
    * Integration tests: Cover empty payload, success scenario, and error
    cases for consumer offset
    operations
    * Payload validation: Added length check in DeserializeOffset to return
    nil instead of panicking on
     empty payloads
---
 .../tests/tcp_test/offset_feature_deserialize.go   | 218 +++++++++++++++++++++
 .../binary_response_deserializer.go                |   4 +
 2 files changed, 222 insertions(+)

diff --git a/bdd/go/tests/tcp_test/offset_feature_deserialize.go b/bdd/go/tests/tcp_test/offset_feature_deserialize.go
new file mode 100644
index 00000000..087fbaf7
--- /dev/null
+++ b/bdd/go/tests/tcp_test/offset_feature_deserialize.go
@@ -0,0 +1,218 @@
+// 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 tcp_test
+
+import (
+	iggcon "github.com/apache/iggy/foreign/go/contracts"
+	"github.com/onsi/ginkgo/v2"
+	"github.com/onsi/gomega"
+)
+
+var _ = ginkgo.Describe("GET CONSUMER OFFSET:", func() {
+	prefix := "GetConsumerOffset"
+
+	ginkgo.When("User is logged in", func() {
+		ginkgo.Context("and tries to get offset for existing consumer group", func() {
+			client := createAuthorizedConnection()
+			streamId, _ := successfullyCreateStream(prefix, client)
+			defer deleteStreamAfterTests(streamId, client)
+			topicId, _ := successfullyCreateTopic(streamId, client)
+			groupId, _ := successfullyCreateConsumer(streamId, topicId, client)
+
+			streamIdentifier, _ := iggcon.NewIdentifier(streamId)
+			topicIdentifier, _ := iggcon.NewIdentifier(topicId)
+			groupIdentifier, _ := iggcon.NewIdentifier(groupId)
+
+			joinErr := client.JoinConsumerGroup(
+				streamIdentifier,
+				topicIdentifier,
+				groupIdentifier,
+			)
+
+			consumer := iggcon.NewGroupConsumer(groupIdentifier)
+			partitionId := uint32(1)
+
+			offset, err := client.GetConsumerOffset(
+				consumer,
+				streamIdentifier,
+				topicIdentifier,
+				&partitionId,
+			)
+
+			itShouldNotReturnError(joinErr)
+			itShouldNotReturnError(err)
+			itShouldReturnNilOffsetForNewConsumerGroup(offset)
+		})
+
+		ginkgo.Context("and gets valid offset after sending messages and storing offset", func() {
+			client := createAuthorizedConnection()
+			streamId, _ := successfullyCreateStream(prefix+"Success", client)
+			defer deleteStreamAfterTests(streamId, client)
+			topicId, _ := successfullyCreateTopic(streamId, client)
+			groupId, _ := successfullyCreateConsumer(streamId, topicId, client)
+
+			streamIdentifier, _ := iggcon.NewIdentifier(streamId)
+			topicIdentifier, _ := iggcon.NewIdentifier(topicId)
+			consumerIdentifier, _ := iggcon.NewIdentifier(groupId)
+
+			partitionId := uint32(1)
+			testOffset := uint64(1)
+
+			consumer := iggcon.NewSingleConsumer(consumerIdentifier)
+
+			messages := createDefaultMessages()
+			sendErr := client.SendMessages(
+				streamIdentifier,
+				topicIdentifier,
+				iggcon.PartitionId(partitionId),
+				messages,
+			)
+
+			storeErr := client.StoreConsumerOffset(
+				consumer,
+				streamIdentifier,
+				topicIdentifier,
+				testOffset,
+				&partitionId,
+			)
+
+			offset, getErr := client.GetConsumerOffset(
+				consumer,
+				streamIdentifier,
+				topicIdentifier,
+				&partitionId,
+			)
+
+			itShouldNotReturnError(sendErr)
+			itShouldNotReturnError(storeErr)
+			itShouldNotReturnError(getErr)
+			itShouldReturnStoredConsumerOffset(offset, partitionId, testOffset)
+		})
+
+		ginkgo.Context("and tries to store and retrieve consumer offset", func() {
+			client := createAuthorizedConnection()
+			streamId, _ := successfullyCreateStream(prefix+"Store", client)
+			defer deleteStreamAfterTests(streamId, client)
+			topicId, _ := successfullyCreateTopic(streamId, client)
+			groupId, _ := successfullyCreateConsumer(streamId, topicId, client)
+
+			streamIdentifier, _ := iggcon.NewIdentifier(streamId)
+			topicIdentifier, _ := iggcon.NewIdentifier(topicId)
+			groupIdentifier, _ := iggcon.NewIdentifier(groupId)
+
+			joinErr := client.JoinConsumerGroup(
+				streamIdentifier,
+				topicIdentifier,
+				groupIdentifier,
+			)
+
+			consumer := iggcon.NewGroupConsumer(groupIdentifier)
+			partitionId := uint32(1)
+			testOffset := uint64(0)
+
+			_ = client.StoreConsumerOffset(
+				consumer,
+				streamIdentifier,
+				topicIdentifier,
+				testOffset,
+				&partitionId,
+			)
+
+			storedOffset, getErr := client.GetConsumerOffset(
+				consumer,
+				streamIdentifier,
+				topicIdentifier,
+				&partitionId,
+			)
+
+			itShouldNotReturnError(joinErr)
+			itShouldNotReturnError(getErr)
+			itShouldReturnNilOffsetForNewConsumerGroup(storedOffset)
+		})
+
+		ginkgo.Context("and tries to get offset from non-existing consumer group", func() {
+			client := createAuthorizedConnection()
+			streamId, _ := successfullyCreateStream(prefix, client)
+			defer deleteStreamAfterTests(streamId, client)
+			topicId, _ := successfullyCreateTopic(streamId, client)
+
+			streamIdentifier, _ := iggcon.NewIdentifier(streamId)
+			topicIdentifier, _ := iggcon.NewIdentifier(topicId)
+			consumer := iggcon.NewGroupConsumer(randomU32Identifier())
+			partitionId := uint32(1)
+
+			offset, err := client.GetConsumerOffset(
+				consumer,
+				streamIdentifier,
+				topicIdentifier,
+				&partitionId,
+			)
+
+			itShouldNotReturnError(err)
+			itShouldReturnNilOffsetForNewConsumerGroup(offset)
+		})
+
+		ginkgo.Context("and tries to get offset from non-existing stream", func() {
+			client := createAuthorizedConnection()
+			consumer := iggcon.NewGroupConsumer(randomU32Identifier())
+			partitionId := uint32(1)
+
+			offset, err := client.GetConsumerOffset(
+				consumer,
+				randomU32Identifier(),
+				randomU32Identifier(),
+				&partitionId,
+			)
+
+			itShouldNotReturnError(err)
+			itShouldReturnNilOffsetForNewConsumerGroup(offset)
+		})
+	})
+
+	ginkgo.When("User is not logged in", func() {
+		ginkgo.Context("and tries to get consumer offset", func() {
+			client := createClient()
+			consumer := iggcon.NewGroupConsumer(randomU32Identifier())
+			partitionId := uint32(1)
+
+			offset, err := client.GetConsumerOffset(
+				consumer,
+				randomU32Identifier(),
+				randomU32Identifier(),
+				&partitionId,
+			)
+
+			itShouldNotReturnError(err)
+			itShouldReturnNilOffsetForNewConsumerGroup(offset)
+		})
+	})
+})
+
+func itShouldReturnNilOffsetForNewConsumerGroup(offset *iggcon.ConsumerOffsetInfo) {
+	ginkgo.It("should return nil offset for new consumer group with no stored offset", func() {
+		gomega.Expect(offset).To(gomega.BeNil(), "Offset should be nil for new consumer group")
+	})
+}
+
+func itShouldReturnStoredConsumerOffset(offset *iggcon.ConsumerOffsetInfo, expectedPartitionId uint32, expectedStoredOffset uint64) {
+	ginkgo.It("should return the stored consumer offset", func() {
+		gomega.Expect(offset).NotTo(gomega.BeNil(), "Offset should not be nil")
+		gomega.Expect(offset.PartitionId).To(gomega.Equal(expectedPartitionId), "PartitionId should match")
+		gomega.Expect(offset.StoredOffset).To(gomega.Equal(expectedStoredOffset), "StoredOffset should match the value we set")
+	})
+}
diff --git a/foreign/go/binary_serialization/binary_response_deserializer.go b/foreign/go/binary_serialization/binary_response_deserializer.go
index c7490f57..bf9437cb 100644
--- a/foreign/go/binary_serialization/binary_response_deserializer.go
+++ b/foreign/go/binary_serialization/binary_response_deserializer.go
@@ -37,6 +37,10 @@ func DeserializeLogInResponse(payload []byte) *iggcon.IdentityInfo {
 }
 
 func DeserializeOffset(payload []byte) *iggcon.ConsumerOffsetInfo {
+	if len(payload) == 0 {
+		return nil
+	}
+
 	partitionId := binary.LittleEndian.Uint32(payload[0:4])
 	currentOffset := binary.LittleEndian.Uint64(payload[4:12])
 	storedOffset := binary.LittleEndian.Uint64(payload[12:20])