From c6771d64a0874cb0ed49b33d3a5060aa3a43810d Mon Sep 17 00:00:00 2001
From: Ioannis Simeonidis <36934072+simioa@users.noreply.github.com>
Date: Thu, 19 Dec 2024 17:36:02 +0100
Subject: [PATCH] [filebeat][netflow]: fix template sharing (#42079)

Pass the share_templates configuration option into the NetflowV9Protocol
struct. The parameter was not being set, and therefore was always false so
it was not possible to use this option.

Added a test case to prevent future regressions.

Closes #42080

(cherry picked from commit 323c69eb7aa9887bfdd2cb68191a77283be399ad)
---
 CHANGELOG.next.asciidoc                       |  1 +
 .../input/netflow/decoder/config/config.go    |  5 ++
 .../filebeat/input/netflow/decoder/v9/v9.go   | 15 +++---
 .../input/netflow/decoder/v9/v9_test.go       | 50 +++++++++++++++++++
 4 files changed, 64 insertions(+), 7 deletions(-)

diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
index 1ae82246f772..b5354656a938 100644
--- a/CHANGELOG.next.asciidoc
+++ b/CHANGELOG.next.asciidoc
@@ -108,6 +108,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Upgrade github.com/hashicorp/go-retryablehttp to mitigate CVE-2024-6104 {pull}40036[40036]
 - Fix streaming input handling of invalid or empty websocket messages. {pull}42036[42036]
 - Fix awss3 document ID construction when using the CSV decoder. {pull}42019[42019]
+- Fix Netflow Template Sharing configuration handling. {pull}42080[42080]
 
 *Heartbeat*
 
diff --git a/x-pack/filebeat/input/netflow/decoder/config/config.go b/x-pack/filebeat/input/netflow/decoder/config/config.go
index 84f1e050c1fe..5297f3c4a31e 100644
--- a/x-pack/filebeat/input/netflow/decoder/config/config.go
+++ b/x-pack/filebeat/input/netflow/decoder/config/config.go
@@ -136,6 +136,11 @@ func (c *Config) SequenceResetEnabled() bool {
 	return c.detectReset
 }
 
+// ShareTemplatesEnabled returns if template sharing is enabled.
+func (c *Config) ShareTemplatesEnabled() bool {
+	return c.sharedTemplates
+}
+
 // Fields returns the configured fields.
 func (c *Config) Fields() fields.FieldDict {
 	if c.fields == nil {
diff --git a/x-pack/filebeat/input/netflow/decoder/v9/v9.go b/x-pack/filebeat/input/netflow/decoder/v9/v9.go
index 2fafe452c623..4e67dde701f3 100644
--- a/x-pack/filebeat/input/netflow/decoder/v9/v9.go
+++ b/x-pack/filebeat/input/netflow/decoder/v9/v9.go
@@ -50,13 +50,14 @@ func New(config config.Config) protocol.Protocol {
 func NewProtocolWithDecoder(decoder Decoder, config config.Config, logger *log.Logger) *NetflowV9Protocol {
 	ctx, cancel := context.WithCancel(context.Background())
 	pd := &NetflowV9Protocol{
-		ctx:         ctx,
-		cancel:      cancel,
-		decoder:     decoder,
-		logger:      logger,
-		Session:     NewSessionMap(logger, config.ActiveSessionsMetric()),
-		timeout:     config.ExpirationTimeout(),
-		detectReset: config.SequenceResetEnabled(),
+		ctx:            ctx,
+		cancel:         cancel,
+		decoder:        decoder,
+		logger:         logger,
+		Session:        NewSessionMap(logger, config.ActiveSessionsMetric()),
+		timeout:        config.ExpirationTimeout(),
+		detectReset:    config.SequenceResetEnabled(),
+		shareTemplates: config.ShareTemplatesEnabled(),
 	}
 
 	if config.Cache() {
diff --git a/x-pack/filebeat/input/netflow/decoder/v9/v9_test.go b/x-pack/filebeat/input/netflow/decoder/v9/v9_test.go
index a98f6150f6aa..67212c1e4084 100644
--- a/x-pack/filebeat/input/netflow/decoder/v9/v9_test.go
+++ b/x-pack/filebeat/input/netflow/decoder/v9/v9_test.go
@@ -249,3 +249,53 @@ func TestCustomFields(t *testing.T) {
 	assert.Contains(t, flows[0].Fields, "customField")
 	assert.Equal(t, flows[0].Fields["customField"], "Hello :)")
 }
+
+func TestSharedTemplates(t *testing.T) {
+	templateAddr := test.MakeAddress(t, "127.0.0.1:12345")
+	flowsAddr := test.MakeAddress(t, "127.0.0.2:21234")
+	templatePacket := []uint16{
+		// Header
+		// Version, Count, Uptime, Ts, SeqNo, Source
+		9, 1, 11, 11, 22, 22, 33, 33, 0, 1234,
+		// Set #1 (template)
+		0, 20, /*len of set*/
+		999, 3, /*len*/
+		1, 4, // Fields
+		2, 4,
+		3, 4,
+	}
+	flowsPacket := []uint16{
+		// Header
+		// Version, Count, Uptime, Ts, SeqNo, Source
+		9, 1, 11, 11, 22, 22, 33, 34, 0, 1234,
+		// Set #1 (template)
+		999, 16, /*len of set*/
+		1, 1,
+		2, 2,
+		3, 3,
+	}
+
+	t.Run("Template sharing enabled", func(t *testing.T) {
+		cfg := config.Defaults()
+		cfg.WithSharedTemplates(true)
+		proto := New(cfg)
+		flows, err := proto.OnPacket(test.MakePacket(templatePacket), templateAddr)
+		assert.NoError(t, err)
+		assert.Empty(t, flows)
+		flows, err = proto.OnPacket(test.MakePacket(flowsPacket), flowsAddr)
+		assert.NoError(t, err)
+		assert.Len(t, flows, 1)
+	})
+
+	t.Run("Template sharing disabled", func(t *testing.T) {
+		cfg := config.Defaults()
+		cfg.WithSharedTemplates(false)
+		proto := New(cfg)
+		flows, err := proto.OnPacket(test.MakePacket(templatePacket), templateAddr)
+		assert.NoError(t, err)
+		assert.Empty(t, flows)
+		flows, err = proto.OnPacket(test.MakePacket(flowsPacket), flowsAddr)
+		assert.NoError(t, err)
+		assert.Empty(t, flows)
+	})
+}