[9.2](backport #47199) [OTel] Add TLS test to Logstash Exporter (#47230)

* [OTel] Add TLS test to Logstash Exporter (#47199)

This commit verifies that key TLS features in Logstash Exporter still work as expected

Add TLS tests
- TLS connections
- Self-signed certificates
- Keypass support

Add E2E TestLogstashExporterProxyURL to ensure proxy_url config is respected
---------

Co-authored-by: Tiago Queiroz <[email protected]>
(cherry picked from commit 55f5403)

Author: mergify[bot]

NOTICE.txt

@@ -14562,6 +14562,36 @@ Contents of probable licence file $GOMODCACHE/github.com/elastic/opentelemetry-c
    limitations under the License.
 
 
+--------------------------------------------------------------------------------
+Dependency : github.com/elastic/pkcs8
+Version: v1.0.0
+Licence type (autodetected): MIT
+--------------------------------------------------------------------------------
+
+Contents of probable licence file $GOMODCACHE/github.com/elastic/[email protected]/LICENSE:
+
+The MIT License (MIT)
+
+Copyright (c) 2014 youmark
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+
 --------------------------------------------------------------------------------
 Dependency : github.com/elastic/sarama
 Version: v1.19.1-0.20250603175145-7672917f26b6
@@ -45872,36 +45902,6 @@ Contents of probable licence file $GOMODCACHE/github.com/elastic/[email protected]
    limitations under the License.
 
 
---------------------------------------------------------------------------------
-Dependency : github.com/elastic/pkcs8
-Version: v1.0.0
-Licence type (autodetected): MIT
---------------------------------------------------------------------------------
-
-Contents of probable licence file $GOMODCACHE/github.com/elastic/[email protected]/LICENSE:
-
-The MIT License (MIT)
-
-Copyright (c) 2014 youmark
-
-Permission is hereby granted, free of charge, to any person obtaining a copy
-of this software and associated documentation files (the "Software"), to deal
-in the Software without restriction, including without limitation the rights
-to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-copies of the Software, and to permit persons to whom the Software is
-furnished to do so, subject to the following conditions:
-
-The above copyright notice and this permission notice shall be included in all
-copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-SOFTWARE.
-
 --------------------------------------------------------------------------------
 Dependency : github.com/elazarl/goproxy
 Version: v0.0.0-20240909085733-6741dbfc16a1

go.mod

@@ -237,6 +237,7 @@ require (
 require (
 	github.com/apache/arrow-go/v18 v18.4.1
 	github.com/cilium/ebpf v0.19.0
+	github.com/elastic/pkcs8 v1.0.0
 	go.opentelemetry.io/collector/client v1.43.0
 	go.opentelemetry.io/collector/component/componenttest v0.137.0
 	go.opentelemetry.io/collector/confmap/xconfmap v0.137.0
@@ -320,7 +321,6 @@ require (
 	github.com/elastic/elastic-transport-go/v8 v8.7.0 // indirect
 	github.com/elastic/go-docappender/v2 v2.11.2 // indirect
 	github.com/elastic/go-windows v1.0.2 // indirect
-	github.com/elastic/pkcs8 v1.0.0 // indirect
 	github.com/elazarl/goproxy v0.0.0-20240909085733-6741dbfc16a1 // indirect
 	github.com/elazarl/goproxy/ext v0.0.0-20240909085733-6741dbfc16a1 // indirect
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect

libbeat/otelbeat/oteltest/beatsauth_test.go

@@ -97,7 +97,7 @@ func TestMTLS(t *testing.T) {
 	}
 
 	// get client certificates paths
-	clientCertificate, clientKey := getClientCerts(t, caCert)
+	clientCertificate, clientKey := GetClientCerts(t, caCert, "")
 
 	// start test server with given server and root certs
 	certPool := x509.NewCertPool()
@@ -113,11 +113,11 @@ func TestMTLS(t *testing.T) {
 
 	inputConfig := `
 receivers:
-  filebeatreceiver:	
+  filebeatreceiver:
     output:
       elasticsearch:
         hosts: {{ .Host }}
-        ssl: 
+        ssl:
           enabled: true
           certificate_authorities:
             - {{ .CACertificate }}
@@ -183,13 +183,13 @@ func TestCATrustedFingerPrint(t *testing.T) {
 
 	inputConfig := `
 receivers:
-  filebeatreceiver:	
+  filebeatreceiver:
     output:
       elasticsearch:
         hosts: {{ .Host }}
-        ssl: 
+        ssl:
           enabled: true
-          ca_trusted_fingerprint: {{ .CATrustedFingerPrint }} 
+          ca_trusted_fingerprint: {{ .CATrustedFingerPrint }}
 `
 
 	var otelConfigBuffer bytes.Buffer
@@ -361,11 +361,11 @@ func TestVerificationMode(t *testing.T) {
 
 			inputConfig := `
 receivers:
-  filebeatreceiver:	
+  filebeatreceiver:
     output:
       elasticsearch:
         hosts: {{ .Host }}
-        ssl: 
+        ssl:
           enabled: true
           certificate_authorities:
             - {{ .CACertificate }}
@@ -457,7 +457,7 @@ func TestProxyHTTP(t *testing.T) {
 				proxytest.WithRequestLog("https", t.Logf)},
 			inputConfig: `
 receivers:
-  filebeatreceiver:	
+  filebeatreceiver:
     output:
       elasticsearch:
         hosts: {{ .Host }}
@@ -490,7 +490,7 @@ receivers:
 				})},
 			inputConfig: `
 receivers:
-  filebeatreceiver:	
+  filebeatreceiver:
     output:
       elasticsearch:
         hosts: {{ .Host }}
@@ -508,7 +508,7 @@ receivers:
 				proxytest.WithRequestLog("https", t.Logf)},
 			inputConfig: `
 receivers:
-  filebeatreceiver:	
+  filebeatreceiver:
     output:
       elasticsearch:
         hosts: {{ .Host }}
@@ -683,40 +683,6 @@ func mustSendLogs(t *testing.T, exporter exporter.Logs, logs plog.Logs) error {
 	return err
 }
 
-// getClientCerts creates client certificates, writes them to a file and return the path of certificate and key
-func getClientCerts(t *testing.T, caCert tls.Certificate) (certificate string, key string) {
-	// create client certificates
-	clientCerts, err := tlscommontest.GenSignedCert(caCert, x509.KeyUsageCertSign, false, "client", []string{"localhost"}, []net.IP{net.IPv4(127, 0, 0, 1)}, false)
-	if err != nil {
-		t.Fatalf("could not generate certificates: %s", err)
-	}
-
-	clientKey, err := x509.MarshalPKCS8PrivateKey(clientCerts.PrivateKey)
-	if err != nil {
-		t.Fatalf("could not marshal private key: %v", err)
-	}
-
-	tempDir := t.TempDir()
-	clientCertPath := filepath.Join(tempDir, "client-cert.pem")
-	clientKeyPath := filepath.Join(tempDir, "client-key.pem")
-
-	if err = os.WriteFile(clientCertPath, pem.EncodeToMemory(&pem.Block{
-		Type:  "CERTIFICATE",
-		Bytes: clientCerts.Leaf.Raw,
-	}), 0o777); err != nil {
-		t.Fatalf("could not write client certificate to file")
-	}
-
-	if err = os.WriteFile(clientKeyPath, pem.EncodeToMemory(&pem.Block{
-		Type:  "RSA PRIVATE KEY",
-		Bytes: clientKey,
-	}), 0o777); err != nil {
-		t.Fatalf("could not write client key to file")
-	}
-
-	return clientCertPath, clientKeyPath
-}
-
 func getTranslatedConf(t *testing.T, input []byte) *confmap.Conf {
 	c := beatconverter.Converter{}
 

libbeat/otelbeat/oteltest/tls_helper.go

@@ -0,0 +1,79 @@
+// 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 oteltest
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/pem"
+	"net"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/elastic/elastic-agent-libs/transport/tlscommontest"
+	"github.com/elastic/pkcs8"
+)
+
+// GetClientCerts creates client certificates, writes them to a file and return the path of certificate and key
+// if passphrase is passed, it is used to encrypt the key file
+func GetClientCerts(t *testing.T, caCert tls.Certificate, passphrase string) (certificate string, key string) {
+	// create client certificates
+	clientCerts, err := tlscommontest.GenSignedCert(caCert, x509.KeyUsageCertSign, false, "client", []string{"localhost"}, []net.IP{net.IPv4(127, 0, 0, 1)}, false)
+	if err != nil {
+		t.Fatalf("could not generate certificates: %s", err)
+	}
+
+	tempDir := t.TempDir()
+	clientCertPath := filepath.Join(tempDir, "client-cert.pem")
+	clientKeyPath := filepath.Join(tempDir, "client-key.pem")
+
+	if passphrase != "" {
+		clientKey, err := pkcs8.MarshalPrivateKey(clientCerts.PrivateKey, []byte(passphrase), pkcs8.DefaultOpts)
+		if err != nil {
+			t.Fatalf("could not marshal private key: %v", err)
+		}
+
+		if err = os.WriteFile(clientKeyPath, pem.EncodeToMemory(&pem.Block{
+			Type:  "ENCRYPTED PRIVATE KEY",
+			Bytes: clientKey,
+		}), 0400); err != nil {
+			t.Fatalf("could not write client key to file")
+		}
+	} else {
+		clientKey, err := x509.MarshalPKCS8PrivateKey(clientCerts.PrivateKey)
+		if err != nil {
+			t.Fatalf("could not marshal private key: %v", err)
+		}
+		if err = os.WriteFile(clientKeyPath, pem.EncodeToMemory(&pem.Block{
+			Type:  "RSA PRIVATE KEY",
+			Bytes: clientKey,
+		}), 0400); err != nil {
+			t.Fatalf("could not write client key to file")
+		}
+	}
+
+	if err = os.WriteFile(clientCertPath, pem.EncodeToMemory(&pem.Block{
+		Type:  "CERTIFICATE",
+		Bytes: clientCerts.Leaf.Raw,
+	}), 0400); err != nil {
+		t.Fatalf("could not write client certificate to file")
+	}
+
+	return clientCertPath, clientKeyPath
+}

x-pack/filebeat/docker-compose.yml

@@ -58,6 +58,14 @@ services:
       - --addr=:8090
       - --config=/files/config.yaml
 
+  socks5-proxy:
+    image: serjs/go-socks5-proxy:latest
+    container_name: socks5-proxy
+    ports:
+      - "1080:1080"
+    environment:
+      PROXY_USER: elastic
+      PROXY_PASSWORD: changeme
   init-logstash:
     image: busybox:latest
     user: "0:0"
@@ -68,6 +76,7 @@ services:
   logstash:
     depends_on:
       - init-logstash
+      - socks5-proxy
     extends:
       file: ${ES_BEATS}/testing/environments/${STACK_ENVIRONMENT}.yml
       service: logstash

x-pack/filebeat/tests/integration/otel_lsexporter_test.go

@@ -149,6 +149,81 @@ processors:
 	compareOutputFiles(t, fbFilePath, otelFilePath, ignoredFields)
 }
 
+// TestLogstashExporterProxyURL verifies that Filebeat OTel mode can send data to Logstash via a SOCKS5 proxy.
+// Filebeat otel mode sends events to "logstash" via a socks5-proxy container running on localhost:1080
+func TestLogstashExporterProxyURL(t *testing.T) {
+	// ensure the size of events is big enough
+	numEvents := 3
+
+	var beatsCfgFile = `
+filebeat.inputs:
+  - type: filestream
+    id: filestream-input-id
+    enabled: true
+    paths:
+      - %s
+output.logstash:
+  hosts:
+    - "logstash:%s"
+  pipelining: 0
+  worker: 1
+  proxy_url: "socks5://elastic:changeme@localhost:1080"
+queue.mem.flush.timeout: 0s
+processors:
+  - add_host_metadata: ~
+  - add_fields:
+      target: ""
+      fields:
+        testcase: %s
+`
+	testCaseName := uuid.Must(uuid.NewV4()).String()
+	events := generateEvents(numEvents)
+
+	// start filebeat in otel mode
+	fbOTel := integration.NewBeat(
+		t,
+		"filebeat-otel",
+		"../../filebeat.test",
+		"otel",
+	)
+
+	inputFilePath := filepath.Join(fbOTel.TempDir(), "event.json")
+	writeEvents(t, inputFilePath, events)
+
+	fbOTel.WriteConfigFile(fmt.Sprintf(beatsCfgFile, inputFilePath, "5055", testCaseName))
+	fbOTel.Start()
+	defer fbOTel.Stop()
+
+	// Nginx endpoint URLs
+	baseURL := "http://localhost:8082"
+	outOTelFileURL := fmt.Sprintf("%s/%s_otel.json", baseURL, testCaseName)
+
+	// Logstash is outputting to a file inside its container, to access
+	// this file we use Nginx to serve the output folder via HTTP
+	// (see docker-compose.yml for Logstash and Nginx configuration).
+	// Wait to ensure the file can be downloaded via HTTP, the file
+	// being available indicates that Filebeat successfully sent data
+	// to Logstash
+	require.EventuallyWithTf(t,
+		func(ct *assert.CollectT) {
+			for _, url := range []string{outOTelFileURL} {
+				checkURLHasContent(ct, url)
+			}
+		},
+		30*time.Second,
+		1*time.Second,
+		"did not find Logstash output file served via Nginx")
+
+	// download files from Nginx
+	otelFilePath := downloadToTestData(t, outOTelFileURL, fmt.Sprintf("%s_otel.json", testCaseName))
+	otelEvents, err := readAllEvents(otelFilePath)
+
+	require.NoError(t, err, "failed to read otel events")
+	require.Equal(t, numEvents, len(otelEvents),
+		"different number of events: sent=%d, get=%d", numEvents, len(otelEvents))
+
+}
+
 func generateEvents(numEvents int) []string {
 	gofakeit.Seed(time.Now().UnixNano())