[8.19](backport #50097) [filebeat][ABS] - Fix CSV decoder JSON escaping in azure-blob-storage input (#50108)

* [filebeat][ABS] - Fix CSV decoder JSON escaping in azure-blob-storage input (#50097)

Fix CSV decoder type references and broken decodeValue in azure-blob-storage input

The azure-blob-storage input's CSV decode path builds JSON via string
concatenation without escaping field values. CSV values containing double
quotes (e.g. RFC 2045 MIME type parameters) produce malformed JSON, causing
downstream ingest pipeline failures. Add a valueDecoder switch case ahead of
the plain decoder case so the CSV decoder's decodeValue() method is used,
which serializes via json.Marshal for correct escaping. Also fix decodeValue()
itself, which previously cleared its state then called decode(), causing a
'decode called before next' error.
---------

Co-authored-by: Shourie Ganguly <shourie.ganguly@elastic.co>

Author: mergify[bot]

changelog/fragments/1776095974-fix-csv-decoder-json-escaping-abs-input.yaml

@@ -0,0 +1,53 @@
+# 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 CSV decoder producing malformed JSON when field values contain double quotes in azure-blob-storage input.
+
+# 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: |
+  The azure-blob-storage input's CSV decode path builds JSON via string
+  concatenation without escaping field values. CSV values containing double
+  quotes (e.g. RFC 2045 MIME type parameters) produce malformed JSON, causing
+  downstream ingest pipeline failures. Add a valueDecoder switch case ahead of
+  the plain decoder case so the CSV decoder's decodeValue() method is used,
+  which serializes via json.Marshal for correct escaping. Also fix decodeValue()
+  itself, which previously cleared its state then called decode(), causing a
+  "decode called before next" error.
+
+# 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/elastic/beats/pull/XXXXX
+
+# 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/elastic/beats/issues/50097

x-pack/filebeat/input/azureblobstorage/decoding_csv.go

@@ -7,6 +7,7 @@ package azureblobstorage
 import (
 	"bytes"
 	"encoding/csv"
+	"encoding/json"
 	"fmt"
 	"io"
 	"slices"
@@ -106,7 +107,7 @@ func (d *csvDecoder) decodeValue() ([]byte, map[string]any, error) {
 		m[n] = d.current[i]
 	}
 	d.current = d.current[:0]
-	b, err := d.decode()
+	b, err := json.Marshal(m)
 	if err != nil {
 		return nil, nil, err
 	}

x-pack/filebeat/input/azureblobstorage/decoding_test.go

@@ -70,6 +70,20 @@ func TestDecoding(t *testing.T) {
 				},
 			},
 		},
+		{
+			name:          "csv_quoted_values",
+			file:          "txn_quoted.csv",
+			content:       "text/csv",
+			numEvents:     2,
+			assertAgainst: "txn_quoted.json",
+			config: decoderConfig{
+				Codec: &codecConfig{
+					CSV: &csvCodecConfig{
+						Enabled: true,
+					},
+				},
+			},
+		},
 	}
 
 	for _, tc := range testCases {

x-pack/filebeat/input/azureblobstorage/job.go

@@ -194,6 +194,22 @@ func (j *job) decode(ctx context.Context, r io.Reader, id string) error {
 	}
 	var evtOffset int64
 	switch dec := dec.(type) {
+	case valueDecoder:
+		defer dec.close()
+
+		for dec.next() {
+			msg, _, err := dec.decodeValue()
+			if err != nil {
+				if errors.Is(err, io.EOF) {
+					return nil
+				}
+				j.status.UpdateStatus(status.Degraded, err.Error())
+				return err
+			}
+			evt := j.createEvent(string(msg), evtOffset)
+			j.publish(evt, !dec.more(), id)
+		}
+
 	case decoder:
 		defer dec.close()
 

x-pack/filebeat/input/azureblobstorage/mock/data_files.go

@@ -262,6 +262,6 @@ var BeatsFilesContainer_multiline_json_gz = []string{
 }
 
 var BeatsFilesContainer_csv = []string{
-	"{\"id\":\"1\",\"name\":\"Alice\",\"email\":\"alice@example.com\",\"status\":\"active\"}",
-	"{\"id\":\"2\",\"name\":\"Bob\",\"email\":\"bob@example.com\",\"status\":\"inactive\"}",
+	"{\"email\":\"alice@example.com\",\"id\":\"1\",\"name\":\"Alice\",\"status\":\"active\"}",
+	"{\"email\":\"bob@example.com\",\"id\":\"2\",\"name\":\"Bob\",\"status\":\"inactive\"}",
 }

x-pack/filebeat/input/azureblobstorage/testdata/txn_quoted.csv

@@ -0,0 +1,3 @@
+date,time,content-type,response-type,host,category
+2020-01-01,00:00:00,"text/plain; charset=""us-ascii""","multipart/mixed; boundary=""gc0p4Jq0M2Yt08jU534c0p""",test.invalid,demo
+2020-01-02,12:30:00,application/json,application/json,example.invalid,demo

x-pack/filebeat/input/azureblobstorage/testdata/txn_quoted.json

@@ -0,0 +1,18 @@
+[
+    {
+        "date": "2020-01-01",
+        "time": "00:00:00",
+        "content-type": "text/plain; charset=\"us-ascii\"",
+        "response-type": "multipart/mixed; boundary=\"gc0p4Jq0M2Yt08jU534c0p\"",
+        "host": "test.invalid",
+        "category": "demo"
+    },
+    {
+        "date": "2020-01-02",
+        "time": "12:30:00",
+        "content-type": "application/json",
+        "response-type": "application/json",
+        "host": "example.invalid",
+        "category": "demo"
+    }
+]