Merge branch 'main' into remote-es

.buildkite/pipeline.yml

@@ -81,6 +81,7 @@ steps:
           image: "docker.elastic.co/cloud-ci/sonarqube/buildkite-scanner:latest"
         command:
           - "buildkite-agent artifact download build/*coverage.out ."
+          - "buildkite-agent artifact download build/test-unit.out.xml ."
           - "/scan-source-code.sh"
         depends_on:
           - step: "unit-test"

changelog/fragments/1699380976-Replace-any-secret-ref-in-input-map.yaml

@@ -0,0 +1,32 @@
+# 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
+
+# Change summary; a 80ish characters long description of the change.
+summary: Replace all secret references in input objects
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+#description:
+
+# Affected component; a word indicating the component this changeset affects.
+component:
+
+# PR URL; optional; the PR number 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/owner/repo/1234
+
+# 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: 3083

internal/pkg/policy/secret.go

@@ -57,27 +57,41 @@ func getPolicyInputsWithSecrets(ctx context.Context, data *model.PolicyData, bul
 	for _, input := range data.Inputs {
 		newInput := make(map[string]interface{})
 		for k, v := range input {
-			// replace secret refs in input stream fields
-			if k == "streams" {
-				if streams, ok := v.([]any); ok {
-					newInput[k] = processStreams(streams, secretValues)
-				}
-				// replace secret refs in input fields
-			} else if ref, ok := input[k].(string); ok {
-				val := replaceSecretRef(ref, secretValues)
-				newInput[k] = val
-			}
-			// if any field was not processed, add back as is
-			if _, ok := newInput[k]; !ok {
-				newInput[k] = v
-			}
+			newInput[k] = replaceAnyRef(v, secretValues)
 		}
 		result = append(result, newInput)
 	}
 	data.SecretReferences = nil
 	return result, nil
 }
 
+// replaceAnyRef is a generic approach to replacing any secret references in the passed item.
+// It will go through any slices or maps and replace any secret references.
+//
+// go's generic parameters are not a good fit for rewriting this method as the typeswitch will not work.
+func replaceAnyRef(ref any, secrets map[string]string) any {
+	var r any
+	switch val := ref.(type) {
+	case string:
+		r = replaceStringRef(val, secrets)
+	case map[string]any:
+		obj := make(map[string]any)
+		for k, v := range val {
+			obj[k] = replaceAnyRef(v, secrets)
+		}
+		r = obj
+	case []any:
+		arr := make([]any, len(val))
+		for i, v := range val {
+			arr[i] = replaceAnyRef(v, secrets)
+		}
+		r = arr
+	default:
+		r = val
+	}
+	return r
+}
+
 type OutputSecret struct {
 	Path []string
 	ID   string
@@ -162,35 +176,8 @@ func processOutputSecret(ctx context.Context, output smap.Map, bulker bulk.Bulk)
 	return nil
 }
 
-func processStreams(streams []any, secretValues map[string]string) []any {
-	newStreams := make([]any, 0)
-	for _, stream := range streams {
-		if streamMap, ok := stream.(map[string]interface{}); ok {
-			newStream := replaceSecretsInStream(streamMap, secretValues)
-			newStreams = append(newStreams, newStream)
-		} else {
-			newStreams = append(newStreams, stream)
-		}
-	}
-	return newStreams
-}
-
-// if field values are secret refs, replace with secret value, otherwise noop
-func replaceSecretsInStream(streamMap map[string]interface{}, secretValues map[string]string) map[string]interface{} {
-	newStream := make(map[string]interface{})
-	for streamKey, streamVal := range streamMap {
-		if streamRef, ok := streamMap[streamKey].(string); ok {
-			replacedVal := replaceSecretRef(streamRef, secretValues)
-			newStream[streamKey] = replacedVal
-		} else {
-			newStream[streamKey] = streamVal
-		}
-	}
-	return newStream
-}
-
-// replace values mathing a secret ref regex, e.g. $co.elastic.secret{<secret ref>} -> <secret value>
-func replaceSecretRef(ref string, secretValues map[string]string) string {
+// replaceStringRef replaces values matching a secret ref regex, e.g. $co.elastic.secret{<secret ref>} -> <secret value>
+func replaceStringRef(ref string, secretValues map[string]string) string {
 	matches := secretRegex.FindStringSubmatch(ref)
 	if len(matches) > 1 {
 		secretRef := matches[1]

internal/pkg/policy/secret_test.go

@@ -15,45 +15,46 @@ import (
 	ftesting "github.com/elastic/fleet-server/v7/internal/pkg/testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
-func TestReplaceSecretRef(t *testing.T) {
+func TestReplaceStringRef(t *testing.T) {
 	secretRefs := map[string]string{
 		"abcd": "value1",
 	}
-	val := replaceSecretRef("$co.elastic.secret{abcd}", secretRefs)
+	val := replaceStringRef("$co.elastic.secret{abcd}", secretRefs)
 	assert.Equal(t, "value1", val)
 }
 
-func TestReplaceSecretRefPartial(t *testing.T) {
+func TestReplaceStringRefPartial(t *testing.T) {
 	secretRefs := map[string]string{
 		"abcd": "value1",
 	}
-	val := replaceSecretRef("partial $co.elastic.secret{abcd}", secretRefs)
+	val := replaceStringRef("partial $co.elastic.secret{abcd}", secretRefs)
 	assert.Equal(t, "partial value1", val)
 }
 
-func TestReplaceSecretRefPartial2(t *testing.T) {
+func TestReplaceStringRefPartial2(t *testing.T) {
 	secretRefs := map[string]string{
 		"abcd": "http://localhost",
 	}
-	val := replaceSecretRef("$co.elastic.secret{abcd}/services", secretRefs)
+	val := replaceStringRef("$co.elastic.secret{abcd}/services", secretRefs)
 	assert.Equal(t, "http://localhost/services", val)
 }
 
-func TestReplaceSecretRefNotASecret(t *testing.T) {
+func TestReplaceStringRefNotASecret(t *testing.T) {
 	secretRefs := map[string]string{
 		"abcd": "value1",
 	}
-	val := replaceSecretRef("abcd", secretRefs)
+	val := replaceStringRef("abcd", secretRefs)
 	assert.Equal(t, "abcd", val)
 }
 
-func TestReplaceSecretRefNotFound(t *testing.T) {
+func TestReplaceStringRefNotFound(t *testing.T) {
 	secretRefs := map[string]string{
 		"abcd": "value1",
 	}
-	val := replaceSecretRef("$co.elastic.secret{other}", secretRefs)
+	val := replaceStringRef("$co.elastic.secret{other}", secretRefs)
 	assert.Equal(t, "$co.elastic.secret{other}", val)
 }
 
@@ -107,6 +108,53 @@ func TestGetPolicyInputsWithSecretsAndStreams(t *testing.T) {
 	assert.Nil(t, pData.SecretReferences)
 }
 
+func TestPolicyInputSteamsEmbedded(t *testing.T) {
+	refs := []model.SecretReferencesItems{{ID: "ref1"}}
+	inputs := []map[string]interface{}{
+		{"id": "input1", "streams": []interface{}{
+			map[string]interface{}{
+				"id":  "stream1",
+				"key": "val",
+				"embedded": map[string]interface{}{
+					"embedded-key": "embedded-val",
+					"embedded-arr": []interface{}{
+						map[string]interface{}{
+							"embedded-secret": "$co.elastic.secret{ref1}",
+						},
+					}},
+			},
+		}},
+	}
+
+	pData := model.PolicyData{
+		SecretReferences: refs,
+		Inputs:           inputs,
+	}
+	bulker := ftesting.NewMockBulk()
+	expected := []map[string]interface{}{{
+		"id": "input1",
+		"streams": []interface{}{
+			map[string]interface{}{
+				"id":  "stream1",
+				"key": "val",
+				"embedded": map[string]interface{}{
+					"embedded-key": "embedded-val",
+					"embedded-arr": []interface{}{
+						map[string]interface{}{
+							"embedded-secret": "ref1_value",
+						},
+					}},
+			},
+		}},
+	}
+
+	result, err := getPolicyInputsWithSecrets(context.TODO(), &pData, bulker)
+	require.NoError(t, err)
+
+	assert.Equal(t, expected, result)
+
+}
+
 func TestGetPolicyInputsNoopWhenNoSecrets(t *testing.T) {
 	inputs := []map[string]interface{}{
 		{"id": "input1"},
@@ -148,7 +196,7 @@ func TestProcessOutputSecret(t *testing.T) {
 			name: "Output with secrets",
 			outputJSON: `{
 				"secrets": {
-					"password": {"id": "passwordid"}	
+					"password": {"id": "passwordid"}
 				}
 			}`,
 			expectOutputJSON: `{