replace secret refs (#2863)

* replace secret refs

* fix check, remove TODO

* moved ReadSecrets to bulker, tests

* fix lint, more tests

* added nil check

* added changelog

* fix nil check

* added comments

* added integration test on secrets

* Update internal/pkg/policy/secret.go

Co-authored-by: Josh Dover <[email protected]>

* Update internal/pkg/policy/secret.go

Co-authored-by: Josh Dover <[email protected]>

* rename to getSecretValues, moved out regex

* refactored getPolicyInputsWithSecrets

* added kibana_system to kibana.yml

* create kibana user

* fixed integration test

---------

Co-authored-by: Josh Dover <[email protected]>

changelog/fragments/1691492339-replace-secret-refs.yaml

@@ -0,0 +1,33 @@
+# 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: feature
+
+# Change summary; a 80ish characters long description of the change.
+summary: Replacing secret references with secret values
+
+# 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: Secrets are being stored in a separate index in elasticsearch, and secret references are included in an agent policy.
+  Secret references are replaced with secret values in the agent policy before sending it out to agents on checkin.
+
+# 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: 2863
+
+# 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: 2485

dev-tools/e2e/kibana.yml

@@ -4,6 +4,7 @@ server.ssl.enabled: false
 elasticsearch.hosts: [ "http://elasticsearch:9200" ]
 elasticsearch.serviceAccountToken: "${KIBANA_TOKEN}"
 
+
 xpack.securitySolution.packagerTaskInterval: 1s
 xpack.fleet.agents.enabled: true
 xpack.fleet.agents.fleet_server.hosts: ["https://fleet-server:8220"]

internal/pkg/api/handleCheckin.go

@@ -525,6 +525,14 @@ func processPolicy(ctx context.Context, zlog zerolog.Logger, bulker bulk.Bulk, a
 	// Update only the output fields to avoid duping the whole map
 	fields[outputsProperty] = json.RawMessage(outputRaw)
 
+	// replace agent policy inputs with the processed inputs where the secret references were replaced with the secret values
+	inputsRaw, err := json.Marshal(pp.Inputs)
+	if err != nil {
+		return nil, err
+	}
+
+	fields["inputs"] = json.RawMessage(inputsRaw)
+
 	rewrittenPolicy := struct {
 		Policy map[string]json.RawMessage `json:"policy"`
 	}{fields}

internal/pkg/bulk/engine.go

@@ -63,6 +63,8 @@ type Bulk interface {
 
 	// Accessor used to talk to elastic search direcly bypassing bulk engine
 	Client() *elasticsearch.Client
+
+	ReadSecrets(ctx context.Context, secretIds []string) (map[string]string, error)
 }
 
 const kModBulk = "bulk"
@@ -112,6 +114,20 @@ func (b *Bulker) Client() *elasticsearch.Client {
 	return client
 }
 
+// read secrets one by one as there is no bulk API yet to read them in one request
+func (b *Bulker) ReadSecrets(ctx context.Context, secretIds []string) (map[string]string, error) {
+	result := make(map[string]string)
+	esClient := b.Client()
+	for _, id := range secretIds {
+		val, err := ReadSecret(ctx, esClient, id)
+		if err != nil {
+			return nil, err
+		}
+		result[id] = val
+	}
+	return result, nil
+}
+
 // Stop timer, but don't stall on channel.
 // API doesn't not seem to work as specified.
 func stopTimer(t *time.Timer) {

internal/pkg/bulk/secret.go

@@ -0,0 +1,62 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package bulk
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+
+	"github.com/elastic/go-elasticsearch/v8"
+)
+
+type ExtendedClient struct {
+	*elasticsearch.Client
+	Custom *ExtendedAPI
+}
+
+type ExtendedAPI struct {
+	*elasticsearch.Client
+}
+
+// Read secret values with custom ES API added in Fleet ES plugin, there is no direct access to secrets index
+// GET /_fleet/secret/secretId
+func (c *ExtendedAPI) Read(ctx context.Context, secretID string) (*SecretResponse, error) {
+	req, err := http.NewRequestWithContext(ctx, "GET", "/_fleet/secret/"+secretID, nil)
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json")
+	if err != nil {
+		return nil, err
+	}
+
+	res, err := c.Perform(req)
+	if err != nil {
+		return nil, err
+	}
+	defer res.Body.Close()
+	var secretResp SecretResponse
+
+	err = json.NewDecoder(res.Body).Decode(&secretResp)
+	if err != nil {
+		return nil, err
+	}
+	return &secretResp, nil
+}
+
+type SecretResponse struct {
+	Value string
+}
+
+func ReadSecret(ctx context.Context, client *elasticsearch.Client, secretID string) (string, error) {
+	es := ExtendedClient{Client: client, Custom: &ExtendedAPI{client}}
+	res, err := es.Custom.Read(ctx, secretID)
+	if err != nil {
+		return "", err
+	}
+	if res == nil {
+		return "", nil
+	}
+	return (*res).Value, err
+}

internal/pkg/policy/monitor.go

@@ -261,14 +261,14 @@ func (m *monitorT) loadPolicies(ctx context.Context) error {
 	return m.processPolicies(ctx, policies)
 }
 
-func (m *monitorT) processPolicies(_ context.Context, policies []model.Policy) error {
+func (m *monitorT) processPolicies(ctx context.Context, policies []model.Policy) error {
 	if len(policies) == 0 {
 		return nil
 	}
 
 	latest := m.groupByLatest(policies)
 	for _, policy := range latest {
-		pp, err := NewParsedPolicy(policy)
+		pp, err := NewParsedPolicy(ctx, m.bulker, policy)
 		if err != nil {
 			return err
 		}

internal/pkg/policy/parsed_policy.go

@@ -5,9 +5,11 @@
 package policy
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 
+	"github.com/elastic/fleet-server/v7/internal/pkg/bulk"
 	"github.com/elastic/fleet-server/v7/internal/pkg/model"
 	"github.com/elastic/fleet-server/v7/internal/pkg/smap"
 )
@@ -44,9 +46,10 @@ type ParsedPolicy struct {
 	Roles   RoleMapT
 	Outputs map[string]Output
 	Default ParsedPolicyDefaults
+	Inputs  []map[string]interface{}
 }
 
-func NewParsedPolicy(p model.Policy) (*ParsedPolicy, error) {
+func NewParsedPolicy(ctx context.Context, bulker bulk.Bulk, p model.Policy) (*ParsedPolicy, error) {
 	var err error
 
 	var fields map[string]json.RawMessage
@@ -76,6 +79,10 @@ func NewParsedPolicy(p model.Policy) (*ParsedPolicy, error) {
 	if err != nil {
 		return nil, err
 	}
+	policyInputs, err := getPolicyInputsWithSecrets(ctx, fields, bulker)
+	if err != nil {
+		return nil, err
+	}
 
 	// We are cool and the gang
 	pp := &ParsedPolicy{
@@ -86,6 +93,7 @@ func NewParsedPolicy(p model.Policy) (*ParsedPolicy, error) {
 		Default: ParsedPolicyDefaults{
 			Name: defaultName,
 		},
+		Inputs: policyInputs,
 	}
 
 	return pp, nil

internal/pkg/policy/parsed_policy_test.go

@@ -2,9 +2,12 @@
 // or more contributor license agreements. Licensed under the Elastic License;
 // you may not use this file except in compliance with the Elastic License.
 
+//go:build !integration
+
 package policy
 
 import (
+	"context"
 	"encoding/json"
 	"testing"
 
@@ -27,7 +30,7 @@ func TestNewParsedPolicy(t *testing.T) {
 
 		m.Data = json.RawMessage(testPolicy)
 
-		pp, err := NewParsedPolicy(m)
+		pp, err := NewParsedPolicy(context.TODO(), nil, m)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -83,7 +86,7 @@ func TestNewParsedPolicyNoES(t *testing.T) {
 
 	m.Data = json.RawMessage(logstashOutputPolicy)
 
-	pp, err := NewParsedPolicy(m)
+	pp, err := NewParsedPolicy(context.TODO(), nil, m)
 	if err != nil {
 		t.Fatal(err)
 	}

internal/pkg/policy/policy_output_integration_test.go

@@ -24,6 +24,8 @@ import (
 	ftesting "github.com/elastic/fleet-server/v7/internal/pkg/testing"
 )
 
+var TestPayload []byte
+
 func TestRenderUpdatePainlessScript(t *testing.T) {
 	tts := []struct {
 		name string

internal/pkg/policy/policy_output_test.go

@@ -2,6 +2,8 @@
 // or more contributor license agreements. Licensed under the Elastic License;
 // you may not use this file except in compliance with the Elastic License.
 
+//go:build !integration
+
 package policy
 
 import (

internal/pkg/policy/secret.go

@@ -0,0 +1,132 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package policy
+
+import (
+	"context"
+	"encoding/json"
+	"regexp"
+	"strings"
+
+	"github.com/elastic/fleet-server/v7/internal/pkg/bulk"
+)
+
+type SecretReference struct {
+	ID string `json:"id"`
+}
+
+var (
+	secretRegex = regexp.MustCompile(`\$co\.elastic\.secret{(.*)}`)
+)
+
+// read secret values that belong to the agent policy's secret references, returns secrets as id:value map
+func getSecretValues(ctx context.Context, secretRefsRaw json.RawMessage, bulker bulk.Bulk) (map[string]string, error) {
+	if secretRefsRaw == nil {
+		return nil, nil
+	}
+
+	var secretValues []SecretReference
+	err := json.Unmarshal([]byte(secretRefsRaw), &secretValues)
+	if err != nil {
+		return nil, err
+	}
+
+	ids := make([]string, 0)
+	for _, ref := range secretValues {
+		ids = append(ids, ref.ID)
+	}
+
+	results, err := bulker.ReadSecrets(ctx, ids)
+	if err != nil {
+		return nil, err
+	}
+
+	return results, nil
+}
+
+// read inputs and secret_references from agent policy
+// replace values of secret refs in inputs and input streams properties
+func getPolicyInputsWithSecrets(ctx context.Context, fields map[string]json.RawMessage, bulker bulk.Bulk) ([]map[string]interface{}, error) {
+	if fields["inputs"] == nil {
+		return nil, nil
+	}
+
+	var inputs []map[string]interface{}
+	err := json.Unmarshal([]byte(fields["inputs"]), &inputs)
+	if err != nil {
+		return nil, err
+	}
+
+	if fields["secret_references"] == nil {
+		return inputs, nil
+	}
+
+	secretValues, err := getSecretValues(ctx, fields["secret_references"], bulker)
+	if err != nil {
+		return nil, err
+	}
+
+	result := make([]map[string]interface{}, 0)
+	for _, input := range inputs {
+		newInput := make(map[string]interface{})
+		for k, v := range input {
+			// replace secret refs in input stream fields
+			if k == "streams" {
+				if streams, ok := input[k].([]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
+			}
+		}
+		result = append(result, newInput)
+	}
+	return result, 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 {
+	matches := secretRegex.FindStringSubmatch(ref)
+	if len(matches) > 1 {
+		secretRef := matches[1]
+		if val, ok := secretValues[secretRef]; ok {
+			return strings.Replace(ref, matches[0], val, 1)
+		}
+	}
+	return ref
+}