fix(filebeat): prevent data corruption in TCP/Syslog/Unix input (#47618)

* fix(filebeat): prevent data corruption in TCP/Syslog input

Mutating a buffer received from a TCP callback could corrupt data in the
underlying bufio.Scanner. Copy the data to a separate buffer before
passing it to the callback.

* mention unix in changelog file

* fix imports

Author: mauri870

changelog/fragments/1763040576-fix-possible-data-corruption-in-tcp-syslog-unix-inputs.yaml

@@ -0,0 +1,45 @@
+# REQUIRED
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: Fix possible data corruption in tcp, syslog and unix inputs
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# description:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: filebeat
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+# pr: https://github.com/owner/repo/1234
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+# issue: https://github.com/owner/repo/1234

filebeat/input/net/tcp/input_test.go

@@ -20,18 +20,26 @@
 package tcp
 
 import (
+	"bytes"
 	"context"
 	"errors"
+	"fmt"
+	"net"
+	"runtime"
 	"sync"
 	"testing"
 	"time"
 
 	netinput "github.com/elastic/beats/v7/filebeat/input/net"
 	"github.com/elastic/beats/v7/filebeat/input/net/nettest"
 	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
+	libbeattesting "github.com/elastic/beats/v7/libbeat/testing"
 	conf "github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/monitoring"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestInput(t *testing.T) {
@@ -95,3 +103,77 @@ func TestInput(t *testing.T) {
 		// No more events on the channel, test passed
 	}
 }
+
+func BenchmarkInput(b *testing.B) {
+	port, err := libbeattesting.AvailableTCP4Port()
+	if err != nil {
+		b.Fatalf("cannot find available port: %s", err)
+	}
+	serverAddr := net.JoinHostPort("localhost", fmt.Sprintf("%d", port))
+
+	inp, err := configure(conf.MustNewConfigFrom(map[string]any{
+		"host":              serverAddr,
+		"number_of_workers": 2,
+	}))
+	if err != nil {
+		b.Fatalf("cannot create input: %s", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	v2Ctx := v2.Context{
+		ID:              b.Name(),
+		Cancelation:     ctx,
+		Logger:          logp.NewNopLogger(),
+		MetricsRegistry: monitoring.NewRegistry(),
+	}
+
+	metrics := inp.InitMetrics("tcp", v2Ctx.MetricsRegistry, v2Ctx.Logger)
+	c := make(chan netinput.DataMetadata, 5*runtime.NumCPU())
+
+	go func() {
+		if err := inp.Run(v2Ctx, c, metrics); err != nil {
+			if !errors.Is(err, context.Canceled) {
+				b.Errorf("input exited with error: %s", err)
+			}
+		}
+	}()
+
+	require.EventuallyWithTf(b, func(ct *assert.CollectT) {
+		conn, err := net.Dial("tcp", serverAddr)
+		require.NoError(ct, err)
+		conn.Close()
+	}, 30*time.Second, 100*time.Millisecond, "waiting for TCP server to start")
+
+	testMessage := bytes.Repeat([]byte("A"), 1001)
+	testMessage[len(testMessage)-1] = '\n'
+	b.ResetTimer()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			conn, err := net.Dial("tcp", serverAddr)
+			if err != nil {
+				b.Errorf("cannot create connection: %s", err)
+				continue
+			}
+
+			for range 100 {
+				_, err = conn.Write(testMessage)
+				if err != nil {
+					b.Errorf("failed to send data: %s", err)
+					break
+				}
+			}
+			conn.Close()
+
+			// Read the events from the channel to prevent blocking
+			for range 100 {
+				select {
+				case <-c:
+				case <-time.After(time.Second):
+					b.Error("timeout waiting for event")
+				}
+			}
+		}
+	})
+}

filebeat/inputsource/common/streaming/handler.go

@@ -78,7 +78,11 @@ func SplitHandlerFactory(family inputsource.Family, logger *logp.Logger, metadat
 					return fmt.Errorf(string(family)+" split_client error: %w", err)
 				}
 				r.Reset()
-				callback(scanner.Bytes(), metadata)
+
+				// Deep copy the data to avoid mutating the underlying scanner buffer.
+				buf := make([]byte, len(scanner.Bytes()))
+				_ = copy(buf, scanner.Bytes())
+				callback(buf, metadata)
 			}
 
 			// We are out of the scanner, either we reached EOF or another fatal error occurred.

filebeat/inputsource/common/streaming/handler_test.go

@@ -0,0 +1,102 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. 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 streaming
+
+import (
+	"bufio"
+	"net"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/beats/v7/filebeat/inputsource"
+	"github.com/elastic/beats/v7/libbeat/common/cfgtype"
+	"github.com/elastic/elastic-agent-libs/logp"
+)
+
+// TestSplitHandlerDeepCopy tests that the SplitHandlerFactory creates a handler
+// that provides deep copies of the scanned data to the consumer callback.
+func TestSplitHandlerDeepCopy(t *testing.T) {
+	logger := logp.NewNopLogger()
+
+	var receivedMessages []string
+	var bufferSnapshots [][]byte
+
+	// Simulate a consumer callback:
+	// 1. Stores the received message as a string for verification.
+	// 2. Keeps a reference to the raw byte slice.
+	// 3. Overwrites the slice to detect whether future callbacks receive
+	//    shared memory instead of independent copies.
+	networkFunc := func(data []byte, metadata inputsource.NetworkMetadata) {
+		receivedMessages = append(receivedMessages, string(data))
+		bufferSnapshots = append(bufferSnapshots, data)
+
+		for i := range data {
+			data[i] = 'X'
+		}
+	}
+
+	metadataFunc := func(conn net.Conn) inputsource.NetworkMetadata {
+		return inputsource.NetworkMetadata{}
+	}
+
+	server, client := net.Pipe()
+	defer server.Close()
+	defer client.Close()
+
+	header := "PREFIX:"
+	smallMsg := header + "small_event_data"
+	largeMsg := header + strings.Repeat("large_event_data_that_fills_buffer_", 10) + "end"
+	expectedMessages := []string{smallMsg, largeMsg, smallMsg}
+
+	input := strings.Join(expectedMessages, "\n") + "\n"
+
+	go func() {
+		defer client.Close()
+		_, err := client.Write([]byte(input))
+		require.NoError(t, err)
+	}()
+
+	factory := SplitHandlerFactory(inputsource.FamilyTCP, logger, metadataFunc, networkFunc, bufio.ScanLines)
+	config := ListenerConfig{
+		MaxMessageSize: cfgtype.ByteSize(2048),
+		Timeout:        1 * time.Second,
+	}
+	handler := factory(config)
+
+	err := handler(t.Context(), server)
+	assert.NoError(t, err)
+
+	require.Len(t, receivedMessages, len(expectedMessages), "should receive all messages")
+
+	for i, received := range receivedMessages {
+		assert.Equal(t, expectedMessages[i], received, "message %d should be received intact", i)
+	}
+
+	// Verify that each callback received an independent copy of the buffer.
+	for i := range bufferSnapshots {
+		assert.Equal(t,
+			strings.Repeat("X", len(expectedMessages[i])),
+			string(bufferSnapshots[i]),
+			"buffer %d should be fully mutated inside the callback", i,
+		)
+	}
+}