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

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 @@
 	"time"
 
 	"gopkg.in/natefinch/lumberjack.v2"
+	"gopkg.in/yaml.v3"
 
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/internal/httplog"
 	"github.com/elastic/beats/v7/x-pack/filebeat/otel"
@@ -57,7 +58,7 @@
 	// 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 @@
 	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.
@@ -324,7 +349,7 @@
 		return err
 	}
 	if !ok {
 		return fmt.Errorf("request tracer path must be within %q path", paths.Resolve(paths.Logs, inputName))
 	}
 	return nil
 }

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

@@ -54,6 +54,61 @@ 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("nested_string", func(t *testing.T) {
+		var s secretState
+		err := s.Unpack("headers:\n  auth: token\n  key: secret")
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		headers, ok := s.m["headers"].(map[string]interface{})
+		if !ok {
+			t.Fatalf("expected map[string]interface{} for nested map, got %T", s.m["headers"])
+		}
+		if headers["auth"] != "token" {
+			t.Fatalf("unexpected value: got %v, want %q", headers["auth"], "token")
+		}
+	})
+
+	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

@@ -205,7 +205,7 @@
 			return err
 		}
 		if !ok {
 			return fmt.Errorf("request tracer path %q must be within %q path", path, paths.Resolve(paths.Logs, inputName))
 		}
 		cfg.Resource.Tracer.Filename = resolved
 	}
@@ -262,8 +262,8 @@
 	} 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{}
@@ -1243,7 +1243,7 @@
 }
 
 func (d socketDialer) Dial(_, _ string) (net.Conn, error) {
 	return net.Dial("unix", d.path)
 }
 
 func (d socketDialer) DialContext(ctx context.Context, _, _ string) (net.Conn, error) {
@@ -1577,7 +1577,7 @@
 		return "80"
 	}()
 
 	_, err := net.DialTimeout("tcp", net.JoinHostPort(url.Hostname(), port), time.Second)
 	if err != nil {
 		return fmt.Errorf("url %q is unreachable: %w", url, err)
 	}

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

@@ -44,14 +44,14 @@
 	return []inputcursor.Source{src}, input{}, nil
 }
 
 // checkUnsupportedParams checks if unsupported/deprecated/discouraged paramaters are set and logs a warning
 func (c config) checkUnsupportedParams(logger *logp.Logger) {
 	if c.RecordCoverage {
 		logger.Named("cel").Warn("execution coverage enabled: " +
 			"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 {