x-pack/filebeat/input/cel: accept string values for secret_state

Fleet resolves secrets to their stored string values, so when
secret_state is configured with secret: true in an integration
package, the agent delivers a YAML text string rather than a
parsed map. Change SecretState from map[string]interface{} to a
custom type that implements the ucfg Unpacker interface, accepting
either a map or a string and parsing the string as YAML.

This works around a Fleet bug where type: yaml variables with
secret: true are silently corrupted during policy compilation.
See elastic/kibana#267859

Author: efd6

changelog/fragments/1778038687-cel-secret-state-unpack.yaml

@@ -0,0 +1,45 @@
+# 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: Accept string values for secret_state to support Fleet secret resolution.
+
+# 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:
+
+# 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/owner/repo/1234
+
+# 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/owner/repo/1234

x-pack/filebeat/input/cel/config.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"gopkg.in/natefinch/lumberjack.v2"
+	"gopkg.in/yaml.v2"
 
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/internal/httplog"
 	"github.com/elastic/beats/v7/x-pack/filebeat/otel"
@@ -57,7 +58,7 @@ type config struct {
 	// placed at state.secret before CEL program execution.
 	// The state.secret key is unconditionally redacted in
 	// debug logs.
-	SecretState map[string]interface{} `config:"secret_state"`
+	SecretState secretState `config:"secret_state"`
 	// Redact is the debug log state redaction configuration.
 	Redact *redact `config:"redact"`
 
@@ -100,6 +101,30 @@ func (c config) GetPackageData(key string) string {
 	return value
 }
 
+// secretState holds secret key-value pairs. It implements
+// the ucfg Unpacker interface to accept either a map (from
+// direct config) or a string (from Fleet secret resolution,
+// which delivers the stored YAML text as a scalar).
+//
+// The string case is needed because Fleet resolves secrets
+// to their stored string values. See
+// https://github.com/elastic/kibana/issues/267859
+type secretState struct {
+	m map[string]interface{}
+}
+
+func (s *secretState) Unpack(v interface{}) error {
+	switch v := v.(type) {
+	case map[string]interface{}:
+		s.m = v
+		return nil
+	case string:
+		return yaml.Unmarshal([]byte(v), &s.m)
+	default:
+		return fmt.Errorf("secret_state: expected string or map, got %T", v)
+	}
+}
+
 type redact struct {
 	// Fields indicates which fields to apply redaction to prior
 	// to logging.

x-pack/filebeat/input/cel/config_test.go

@@ -54,6 +54,46 @@ func TestRegexpConfig(t *testing.T) {
 	}
 }
 
+func TestSecretStateUnpack(t *testing.T) {
+	t.Run("map", func(t *testing.T) {
+		var s secretState
+		err := s.Unpack(map[string]interface{}{"api_key": "secret"})
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		if s.m["api_key"] != "secret" {
+			t.Fatalf("unexpected value: got %v, want %q", s.m["api_key"], "secret")
+		}
+	})
+
+	t.Run("string", func(t *testing.T) {
+		var s secretState
+		err := s.Unpack("api_key: secret")
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		if s.m["api_key"] != "secret" {
+			t.Fatalf("unexpected value: got %v, want %q", s.m["api_key"], "secret")
+		}
+	})
+
+	t.Run("invalid_type", func(t *testing.T) {
+		var s secretState
+		err := s.Unpack(42)
+		if err == nil {
+			t.Fatal("expected error for int value")
+		}
+	})
+
+	t.Run("invalid_yaml", func(t *testing.T) {
+		var s secretState
+		err := s.Unpack(":\ninvalid:\n  :\n")
+		if err == nil {
+			t.Fatal("expected error for invalid YAML")
+		}
+	})
+}
+
 func TestSecretStateValidation(t *testing.T) {
 	base := config{
 		Interval: time.Minute,

x-pack/filebeat/input/cel/input.go

@@ -262,8 +262,8 @@ func (i input) run(env v2.Context, src *source, cursor map[string]interface{}, p
 	} else {
 		state = cfg.State
 	}
-	if len(cfg.SecretState) > 0 {
-		state["secret"] = cfg.SecretState
+	if len(cfg.SecretState.m) > 0 {
+		state["secret"] = cfg.SecretState.m
 	}
 	if cfg.Redact == nil {
 		cfg.Redact = &redact{}

x-pack/filebeat/input/cel/input_manager.go

@@ -51,7 +51,7 @@ func (c config) checkUnsupportedParams(logger *logp.Logger) {
 			"see documentation for details: https://www.elastic.co/guide/en/beats/filebeat/current/filebeat-input-cel.html#cel-record-coverage")
 	}
 	if c.Redact == nil {
-		if len(c.SecretState) > 0 {
+		if len(c.SecretState.m) > 0 {
 			logger.Named("cel").Warn("state.secret is automatically redacted, but 'redact' configuration is recommended if other state fields contain sensitive values: " +
 				"see documentation for details: https://www.elastic.co/guide/en/beats/filebeat/current/filebeat-input-cel.html#cel-state-redact")
 		} else {