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

x-pack/filebeat/input/azureblobstorage: Fix CSV decoder JSON escaping in azure-blob-storage input

The azure-blob-storage input's decode path only matched the
decoder.Decoder interface, calling Decode() which 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.

Added a decoder.ValueDecoder switch case ahead of decoder.Decoder,
matching the pattern already used by the GCS input. ValueDecoder's
DecodeValue() uses json.Marshal which handles escaping correctly.

(cherry picked from commit 34634a3)

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,51 @@
+# 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 decode path only matched the decoder.Decoder
+  interface, which 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 decoder.ValueDecoder switch case which uses json.Marshal
+  for correct escaping, matching the pattern already used by the GCS input.
+
+# 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_test.go

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

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

@@ -195,6 +195,22 @@ func (j *job) decode(ctx context.Context, r io.Reader, id string) error {
 	}
 	var evtOffset int64
 	switch dec := dec.(type) {
+	case decoder.ValueDecoder:
+		defer dec.Close()
+
+		for dec.Next() {
+			offset, 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), offset)
+			j.publish(evt, !dec.More(), id)
+		}
+
 	case decoder.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"
+    }
+]