fluentforwardreceiver allocates arbitrary memory from attacker-controlled msgpack length prefixes

.chloggen/49295.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component: receiver/fluent_forward
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Fix a denial-of-service (memory exhaustion) vulnerability in the `fluentforward` receiver by validating msgpack array and map length fields before allocation."
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [49295]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

receiver/fluentforwardreceiver/conversion.go

@@ -18,7 +18,10 @@
 	"go.opentelemetry.io/collector/pdata/plog"
 )
 
-const tagAttributeKey = "fluent.tag"
+const (
+	tagAttributeKey    = "fluent.tag"
+	maxFluentArrayLen  = 1 << 20
+)
 
 // Most of this logic is derived directly from
 // https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1,
@@ -34,6 +37,7 @@
 type optionsMap map[string]any
 
 // Chunk returns the `chunk` option or blank string if it was not set.
+// ... (omitting actual comments for brevity, but let's keep them)
 func (om optionsMap) Chunk() string {
 	c, _ := om["chunk"].(string)
 	return c
@@ -157,6 +161,9 @@
 	if err != nil {
 		return msgp.WrapError(err, "Record")
 	}
+	if recordLen > maxFluentArrayLen {
+		return msgp.WrapError(fmt.Errorf("map length %d exceeds max length %d", recordLen, maxFluentArrayLen), "Record")
+	}
 
 	for recordLen > 0 {
 		recordLen--
@@ -234,6 +241,9 @@
 	if err != nil {
 		return nil, msgp.WrapError(err, "Option")
 	}
+	if optionLen > maxFluentArrayLen {
+		return nil, msgp.WrapError(fmt.Errorf("options map length %d exceeds max length %d", optionLen, maxFluentArrayLen), "Option")
+	}
 	out := make(optionsMap, optionLen)
 
 	for optionLen > 0 {
@@ -280,6 +290,9 @@
 	if err != nil {
 		return msgp.WrapError(err, "Record")
 	}
+	if entryLen > maxFluentArrayLen {
+		return msgp.WrapError(fmt.Errorf("array length %d exceeds max length %d", entryLen, maxFluentArrayLen), "Record")
+	}
 
 	fe.EnsureCapacity(int(entryLen))
 	for i := 0; i < int(entryLen); i++ {

receiver/fluentforwardreceiver/conversion_test.go

@@ -246,3 +246,47 @@
 		},
 	).ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0), le))
 }
+
+func TestDecodeMsgExcessiveLengths(t *testing.T) {
+	t.Run("MessageEventExcessiveRecordMap", func(t *testing.T) {
+		var b []byte
+		b = msgp.AppendArrayHeader(b, 3)
+		b = msgp.AppendString(b, "my-tag")
+		b = msgp.AppendInt(b, 5000)
+		b = msgp.AppendMapHeader(b, maxFluentArrayLen+1)
+
+		reader := msgp.NewReader(bytes.NewReader(b))
+		var event messageEventLogRecord
+		err := event.DecodeMsg(reader)
+		require.ErrorContains(t, err, "exceeds max length")
+	})
+
+	t.Run("MessageEventExcessiveOptionsMap", func(t *testing.T) {
+		var b []byte
+		b = msgp.AppendArrayHeader(b, 4)
+		b = msgp.AppendString(b, "my-tag")
+		b = msgp.AppendInt(b, 5000)
+		b = msgp.AppendMapHeader(b, 1)
+		b = msgp.AppendString(b, "log")
+		b = msgp.AppendString(b, "hello")
+		b = msgp.AppendMapHeader(b, maxFluentArrayLen+1)
+
+		reader := msgp.NewReader(bytes.NewReader(b))
+		var event messageEventLogRecord
+		err := event.DecodeMsg(reader)
+		require.ErrorContains(t, err, "exceeds max length")
+	})
+
+	t.Run("ForwardEventExcessiveEntriesArray", func(t *testing.T) {
+		var b []byte
+		b = msgp.AppendArrayHeader(b, 2)
+		b = msgp.AppendString(b, "my-tag")
+		b = msgp.AppendArrayHeader(b, maxFluentArrayLen+1)
+
+		reader := msgp.NewReader(bytes.NewReader(b))
+		var event forwardEventLogRecords
+		err := event.DecodeMsg(reader)
+		require.ErrorContains(t, err, "exceeds max length")
+	})
+}
+