LOG-5602: Vector forwarding auth issue when password contains $

internal/generator/vector/helpers/helpers.go

@@ -52,3 +52,7 @@ func ListenOnAllLocalInterfacesAddress() string {
 	listenAllOnce.Do(f)
 	return listenAllAddress
 }
+
+func EscapeDollarSigns(s string) string {
+	return strings.ReplaceAll(s, "$", "$$")
+}

internal/generator/vector/output/azuremonitor/azm_common_dollar.toml

@@ -0,0 +1,31 @@
+[transforms.output_azure_monitor_logs_dedot]
+type = "remap"
+inputs = ["pipelineName"]
+source = '''
+  .openshift.sequence = to_unix_timestamp(now(), unit: "nanoseconds")
+  if exists(.kubernetes.namespace_labels) {
+	  for_each(object!(.kubernetes.namespace_labels)) -> |key,value| {
+		newkey = replace(key, r'[\./]', "_")
+		.kubernetes.namespace_labels = set!(.kubernetes.namespace_labels,[newkey],value)
+		if newkey != key {
+		  .kubernetes.namespace_labels = remove!(.kubernetes.namespace_labels,[key],true)
+		}
+	  }
+  }
+  if exists(.kubernetes.labels) {
+	  for_each(object!(.kubernetes.labels)) -> |key,value| {
+		newkey = replace(key, r'[\./]', "_")
+		.kubernetes.labels = set!(.kubernetes.labels,[newkey],value)
+		if newkey != key {
+		  .kubernetes.labels = remove!(.kubernetes.labels,[key],true)
+		}
+	  }
+  }
+'''
+
+[sinks.output_azure_monitor_logs]
+type = "azure_monitor_logs"
+inputs = ["output_azure_monitor_logs_dedot"]
+customer_id = "6vzw6sHc-0bba-6sHc-4b6c-8bz7sr5eggRt"
+log_type = "myLogType"
+shared_key = "mykeyWith$$Sign"

internal/generator/vector/output/azuremonitor/azuremonitor.go

@@ -3,7 +3,6 @@ package azuremonitor
 import (
 	"github.com/openshift/cluster-logging-operator/internal/constants"
 	"github.com/openshift/cluster-logging-operator/internal/generator/framework"
-	"github.com/openshift/cluster-logging-operator/internal/generator/helpers/security"
 	"github.com/openshift/cluster-logging-operator/internal/generator/vector/normalize"
 	"github.com/openshift/cluster-logging-operator/internal/generator/vector/output/common"
 
@@ -73,7 +72,7 @@ func Output(id string, o logging.OutputSpec, inputs []string, secret *corev1.Sec
 		CustomerId:      azm.CustomerId,
 		LogType:         azm.LogType,
 		AzureResourceId: azm.AzureResourceId,
-		SharedKey:       security.GetFromSecret(secret, constants.SharedKey),
+		SharedKey:       common.GetFromSecret(secret, constants.SharedKey),
 		Host:            azm.Host,
 	}
 }

internal/generator/vector/output/azuremonitor/azuremonitor_test.go

@@ -2,9 +2,10 @@ package azuremonitor
 
 import (
 	_ "embed"
+	"testing"
+
 	. "github.com/openshift/cluster-logging-operator/internal/constants"
 	vectorhelpers "github.com/openshift/cluster-logging-operator/internal/generator/vector/helpers"
-	"testing"
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/ginkgo/extensions/table"
@@ -15,19 +16,24 @@ import (
 )
 
 const (
-	sharedKey     = "z9ndQSFH1RLDnS6WR35m84u326p3"
-	azureId       = "/subscriptions/11111111-1111-1111-1111-111111111111/resourceGroups/otherResourceGroup/providers/Microsoft.Storage/storageAccounts/examplestorage"
-	hostCN        = "ods.opinsights.azure.cn"
-	customerId    = "6vzw6sHc-0bba-6sHc-4b6c-8bz7sr5eggRt"
-	secretName    = "azure-monitor-secret"
-	secretTlsName = "azure-monitor-secret-tls"
-	outputName    = "azure_monitor_logs"
-	logType       = "myLogType"
+	sharedKey           = "z9ndQSFH1RLDnS6WR35m84u326p3"
+	sharedKeyWithDollar = "mykeyWith$Sign"
+	azureId             = "/subscriptions/11111111-1111-1111-1111-111111111111/resourceGroups/otherResourceGroup/providers/Microsoft.Storage/storageAccounts/examplestorage"
+	hostCN              = "ods.opinsights.azure.cn"
+	customerId          = "6vzw6sHc-0bba-6sHc-4b6c-8bz7sr5eggRt"
+	secretName          = "azure-monitor-secret"
+	secretTlsName       = "azure-monitor-secret-tls"
+	outputName          = "azure_monitor_logs"
+	secretWithDollar    = "azure-monitor-secret-dollar"
+	logType             = "myLogType"
 )
 
 //go:embed azm_common.toml
 var ExpectedAzureCommonToml string
 
+//go:embed azm_common_dollar.toml
+var ExpectedAzureCommonDollarToml string
+
 //go:embed azm_advance.toml
 var ExpectedAzureAdvanceToml string
 
@@ -54,6 +60,20 @@ var _ = Describe("Generating vector config for Azure Monitor Logs output:", func
 			},
 		}
 
+		outputCommonWithDollar = loggingv1.OutputSpec{
+			Type: loggingv1.OutputTypeAzureMonitor,
+			Name: outputName,
+			OutputTypeSpec: loggingv1.OutputTypeSpec{
+				AzureMonitor: &loggingv1.AzureMonitor{
+					CustomerId: customerId,
+					LogType:    logType,
+				},
+			},
+			Secret: &loggingv1.OutputSecretSpec{
+				Name: secretWithDollar,
+			},
+		}
+
 		outputCommonTls = loggingv1.OutputSpec{
 			Type: loggingv1.OutputTypeAzureMonitor,
 			Name: outputName,
@@ -116,6 +136,11 @@ var _ = Describe("Generating vector config for Azure Monitor Logs output:", func
 					Passphrase:         []byte("foo"),
 				},
 			},
+			secretWithDollar: {
+				Data: map[string][]byte{
+					SharedKey: []byte(sharedKeyWithDollar),
+				},
+			},
 		}
 	)
 
@@ -127,6 +152,7 @@ var _ = Describe("Generating vector config for Azure Monitor Logs output:", func
 		Entry("for advance case", outputAdvance, secretName, ExpectedAzureAdvanceToml),
 		Entry("for common with tls case", outputCommonTls, secretTlsName, ExpectedAzureTlsToml),
 		Entry("for common with skip tls verify true case", outputWithTlsSkipVerify, secretName, ExpectedAzureSkipTls),
+		Entry("for common case with sharedKey containing `$`", outputCommonWithDollar, secretWithDollar, ExpectedAzureCommonDollarToml),
 	)
 })
 

internal/generator/vector/output/common/security.go

@@ -8,7 +8,6 @@ import (
 	logging "github.com/openshift/cluster-logging-operator/api/logging/v1"
 	"github.com/openshift/cluster-logging-operator/internal/constants"
 	"github.com/openshift/cluster-logging-operator/internal/generator/framework"
-	"github.com/openshift/cluster-logging-operator/internal/generator/helpers/security"
 	urlhelper "github.com/openshift/cluster-logging-operator/internal/generator/url"
 	"github.com/openshift/cluster-logging-operator/internal/generator/vector/helpers"
 
@@ -88,7 +87,7 @@ func addTLSSettings(o logging.OutputSpec, secret *corev1.Secret, conf *TLSConf)
 
 	if HasPassphrase(secret) {
 		addTLS = true
-		conf.PassPhrase = security.GetFromSecret(secret, constants.Passphrase)
+		conf.PassPhrase = GetFromSecret(secret, constants.Passphrase)
 	}
 	if conf.TlsMinVersion != "" || conf.CipherSuites != "" {
 		addTLS = true
@@ -206,7 +205,7 @@ func TryKeys(secret *corev1.Secret, keys ...string) (data []byte, ok bool) {
 
 func GetFromSecret(secret *corev1.Secret, name string) string {
 	if secret != nil {
-		return string(secret.Data[name])
+		return helpers.EscapeDollarSigns(string(secret.Data[name]))
 	}
 	return ""
 }

internal/generator/vector/output/elasticsearch/elasticsearch_test.go

@@ -1,10 +1,11 @@
 package elasticsearch
 
 import (
+	"testing"
+
 	"github.com/openshift/cluster-logging-operator/internal/generator/framework"
 	vectorhelpers "github.com/openshift/cluster-logging-operator/internal/generator/vector/helpers"
 	"github.com/openshift/cluster-logging-operator/internal/generator/vector/output/common"
-	"testing"
 
 	"github.com/openshift/cluster-logging-operator/test/helpers"
 
@@ -42,7 +43,7 @@ var _ = Describe("Generate Vector config", func() {
 				"es-1": {
 					Data: map[string][]byte{
 						"username": []byte("testuser"),
-						"password": []byte("testpass"),
+						"password": []byte("test$pass"),
 					},
 				},
 			},
@@ -128,7 +129,7 @@ timeout_secs = 2147483648
 [sinks.es_1.auth]
 strategy = "basic"
 user = "testuser"
-password = "testpass"
+password = "test$$pass"
 `,
 		}),
 		Entry("with tls key,cert,ca-bundle", helpers.ConfGenerateTest{

internal/generator/vector/output/http/http_test.go

@@ -1,10 +1,11 @@
 package http
 
 import (
+	"testing"
+
 	"github.com/openshift/cluster-logging-operator/internal/constants"
 	"github.com/openshift/cluster-logging-operator/internal/generator/framework"
 	"github.com/openshift/cluster-logging-operator/internal/generator/vector/helpers"
-	"testing"
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/ginkgo/extensions/table"
@@ -41,7 +42,7 @@ var _ = Describe("Generate vector config", func() {
 			&corev1.Secret{
 				Data: map[string][]byte{
 					"username": []byte("username"),
-					"password": []byte("password"),
+					"password": []byte("pa$sword"),
 				},
 			},
 			framework.NoOptions,
@@ -94,7 +95,7 @@ headers = {"h1"="v1","h2"="v2"}
 [sinks.http_receiver.auth]
 strategy = "basic"
 user = "username"
-password = "password"
+password = "pa$$sword"
 `,
 		),
 		Entry("with custom bearer token",

internal/generator/vector/output/kafka/kafka_test.go

@@ -1,10 +1,11 @@
 package kafka
 
 import (
+	"testing"
+
 	"github.com/openshift/cluster-logging-operator/internal/generator/framework"
 	"github.com/openshift/cluster-logging-operator/internal/generator/helpers/security"
 	vectorhelpers "github.com/openshift/cluster-logging-operator/internal/generator/vector/helpers"
-	"testing"
 
 	"github.com/openshift/cluster-logging-operator/test/helpers"
 
@@ -43,7 +44,7 @@ var _ = Describe("Generate vector config", func() {
 					Data: map[string][]byte{
 						"sasl.enable": []byte("true"),
 						"username":    []byte("testuser"),
-						"password":    []byte("testpass"),
+						"password":    []byte("test$pass"),
 					},
 				},
 			},
@@ -88,7 +89,7 @@ timestamp_format = "rfc3339"
 [sinks.kafka_receiver.sasl]
 enabled = true
 username = "testuser"
-password = "testpass"
+password = "test$$pass"
 mechanism = "PLAIN"
 `,
 		}),

internal/generator/vector/output/loki/loki_conf_test.go

@@ -125,7 +125,7 @@ var _ = Describe("Generate vector config", func() {
 				"loki-receiver": {
 					Data: map[string][]byte{
 						"username": []byte("username"),
-						"password": []byte("password"),
+						"password": []byte("pa$sword"),
 					},
 				},
 			},

internal/generator/vector/output/loki/with_custom_labels.toml

@@ -66,4 +66,4 @@ kubernetes_labels_app = "{{kubernetes.labels.\"app\"}}"
 [sinks.loki_receiver.auth]
 strategy = "basic"
 user = "username"
-password = "password"
+password = "pa$$sword"

internal/generator/vector/output/splunk/splunk_test.go

@@ -13,8 +13,12 @@ import (
 	corev1 "k8s.io/api/core/v1"
 )
 
-// #nosec G101
-const hecToken = "VS0BNth3wCGF0eol0MuK07SHIrhYwCPHFWMG"
+const (
+	// #nosec G101
+	expHecToken = "VS0BNth3wCGF0eol0MuK07SHIrhY$$wCPHFWMG"
+	// #nosec G101
+	hecToken = "VS0BNth3wCGF0eol0MuK07SHIrhY$wCPHFWMG"
+)
 
 var _ = Describe("Generating vector config for Splunk output", func() {
 
@@ -67,7 +71,7 @@ type = "splunk_hec_logs"
 inputs = ["splunk_hec_timestamp"]
 endpoint = "https://splunk-web:8088/endpoint"
 compression = "none"
-default_token = "` + hecToken + `"
+default_token = "` + expHecToken + `"
 timestamp_key = "@timestamp"
 [sinks.splunk_hec.encoding]
 codec = "json"
@@ -78,7 +82,7 @@ type = "splunk_hec_logs"
 inputs = ["splunk_hec_timestamp"]
 endpoint = "https://splunk-web:8088/endpoint"
 compression = "none"
-default_token = "` + hecToken + `"
+default_token = "` + expHecToken + `"
 timestamp_key = "@timestamp"
 [sinks.splunk_hec.encoding]
 codec = "json"
@@ -94,7 +98,7 @@ type = "splunk_hec_logs"
 inputs = ["splunk_hec_timestamp"]
 endpoint = "https://splunk-web:8088/endpoint"
 compression = "none"
-default_token = "` + hecToken + `"
+default_token = "` + expHecToken + `"
 timestamp_key = "@timestamp"
 [sinks.splunk_hec.encoding]
 codec = "json"
@@ -127,7 +131,7 @@ type = "splunk_hec_logs"
 inputs = ["splunk_hec_timestamp"]
 endpoint = "https://splunk-web:8088/endpoint"
 compression = "none"
-default_token = "` + hecToken + `"
+default_token = "` + expHecToken + `"
 timestamp_key = "@timestamp"
 
 [sinks.splunk_hec.encoding]
@@ -324,7 +328,7 @@ type = "splunk_hec_logs"
 inputs = ["splunk_hec_timestamp"]
 endpoint = "https://splunk-web:8088/endpoint"
 compression = "none"
-default_token = "` + hecToken + `"
+default_token = "` + expHecToken + `"
 index = "{{ write_index }}"
 timestamp_key = "@timestamp"
 [sinks.splunk_hec.encoding]

internal/generator/vector/output/syslog/syslog.go

@@ -159,7 +159,7 @@ func Facility(s *logging.Syslog) string {
 	if IsKeyExpr(s.Facility) {
 		return fmt.Sprintf("$%s", s.Facility)
 	}
-	return s.Facility
+	return vectorhelpers.EscapeDollarSigns(s.Facility)
 }
 
 func Severity(s *logging.Syslog) string {
@@ -169,7 +169,7 @@ func Severity(s *logging.Syslog) string {
 	if IsKeyExpr(s.Severity) {
 		return fmt.Sprintf("$%s", s.Severity)
 	}
-	return s.Severity
+	return vectorhelpers.EscapeDollarSigns(s.Severity)
 }
 
 func RFC(s *logging.Syslog) string {
@@ -199,7 +199,7 @@ func AppName(s *logging.Syslog) Element {
 	if s.AppName == "tag" {
 		return KV(appname, "${tag}")
 	}
-	return KV(appname, fmt.Sprintf(`"%s"`, s.AppName))
+	return KV(appname, fmt.Sprintf(`"%s"`, vectorhelpers.EscapeDollarSigns(s.AppName)))
 }
 
 func Tag(s *logging.Syslog) Element {
@@ -210,7 +210,7 @@ func Tag(s *logging.Syslog) Element {
 	if IsKeyExpr(s.Tag) {
 		return KV(tag, fmt.Sprintf(`"$%s"`, s.Tag))
 	}
-	return KV(tag, fmt.Sprintf(`"%s"`, s.Tag))
+	return KV(tag, fmt.Sprintf(`"%s"`, vectorhelpers.EscapeDollarSigns(s.Tag)))
 }
 
 func MsgID(s *logging.Syslog) Element {
@@ -221,7 +221,7 @@ func MsgID(s *logging.Syslog) Element {
 	if IsKeyExpr(s.MsgID) {
 		return KV(msgid, fmt.Sprintf(`"$%s"`, s.MsgID))
 	}
-	return KV(msgid, fmt.Sprintf(`"%s"`, s.MsgID))
+	return KV(msgid, fmt.Sprintf(`"%s"`, vectorhelpers.EscapeDollarSigns(s.MsgID)))
 }
 
 func ProcID(s *logging.Syslog) Element {
@@ -232,7 +232,7 @@ func ProcID(s *logging.Syslog) Element {
 	if IsKeyExpr(s.ProcID) {
 		return KV(procid, fmt.Sprintf(`"$%s"`, s.ProcID))
 	}
-	return KV(procid, fmt.Sprintf(`"%s"`, s.ProcID))
+	return KV(procid, fmt.Sprintf(`"%s"`, vectorhelpers.EscapeDollarSigns(s.ProcID)))
 }
 
 func AddLogSource(s *logging.Syslog) Element {