x-pack/filebeat/input/cel: add envvar support (#40779)

CHANGELOG.next.asciidoc

@@ -304,6 +304,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Add `use_kubeadm` config option for filebeat (both filbeat.input and autodiscovery) in order to toggle kubeadm-config api requests {pull}40301[40301]
 - Make HTTP library function inclusion non-conditional in CEL input. {pull}40912[40912]
 - Add support for Crowdstrike streaming API to the streaming input. {issue}40264[40264] {pull}40838[40838]
+- Add support to CEL for reading host environment variables. {issue}40762[40762] {pull}40779[40779]
 
 *Auditbeat*
 

x-pack/filebeat/docs/inputs/input-cel.asciidoc

@@ -245,6 +245,8 @@ As noted above the `cel` input provides functions, macros, and global variables
 
 In addition to the extensions provided in the packages listed above, a global variable `useragent` is also provided which gives the user CEL program access to the {beatname_lc} user-agent string. By default, this value is assigned to all requests' user-agent headers unless the CEL program has already set the user-agent header value. Programs wishing to not provide a user-agent, should set this header to the empty string, `""`.
 
+Host environment variables are made available via the global map `env`. Only environment variables that have been allow listed via the `allowed_environment` configuration list are visible to the CEL program.
+
 The CEL environment enables the https://pkg.go.dev/github.com/google/cel-go/cel#OptionalTypes[optional types] library using the version defined {mito_docs}/lib#OptionalTypesVersion[here].
 
 Additionally, it supports authentication via Basic Authentication, Digest Authentication or OAuth2.
@@ -357,6 +359,33 @@ filebeat.inputs:
     })
 ----
 
+[[environ-cel]]
+[float]
+=== `allowed_environment`
+
+A list of host environment variable that will be made visible to the CEL execution environment. By default, no environment variables are visible.
+
+["source","yaml",subs="attributes"]
+----
+filebeat.inputs:
+# Publish the list of files in $PATH every minute.
+- type: cel
+  interval: 1m
+  resource.url: ""
+  allowed_environment:
+    - PATH
+  program: |
+{
+  "events": {
+    "message": env.?PATH.orValue("").split(":")
+      .map(p, try(dir(p)))
+      .filter(d, type(d) != type(""))
+      .flatten()
+      .collate("name")
+  }
+}
+----
+
 [[regexp-cel]]
 [float]
 ==== `regexp`

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

@@ -47,6 +47,10 @@ type config struct {
 	// Redact is the debug log state redaction configuration.
 	Redact *redact `config:"redact"`
 
+	// AllowedEnvironment is the set of env vars made
+	// visible to an executing CEL evaluation.
+	AllowedEnvironment []string `config:"allowed_environment"`
+
 	// Auth is the authentication config for connection to an HTTP
 	// API endpoint.
 	Auth authConfig `config:"auth"`
@@ -85,7 +89,7 @@ func (c config) Validate() error {
 	if len(c.Regexps) != 0 {
 		patterns = map[string]*regexp.Regexp{".": nil}
 	}
-	_, _, err = newProgram(context.Background(), c.Program, root, &http.Client{}, nil, nil, patterns, c.XSDs, logp.L().Named("input.cel"), nil)
+	_, _, err = newProgram(context.Background(), c.Program, root, nil, &http.Client{}, nil, nil, patterns, c.XSDs, logp.L().Named("input.cel"), nil)
 	if err != nil {
 		return fmt.Errorf("failed to check program: %w", err)
 	}

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

@@ -21,6 +21,7 @@ import (
 	"path/filepath"
 	"reflect"
 	"regexp"
+	"slices"
 	"strconv"
 	"strings"
 	"time"
@@ -165,7 +166,7 @@ func (i input) run(env v2.Context, src *source, cursor map[string]interface{}, p
 			Password: cfg.Auth.Basic.Password,
 		}
 	}
-	prg, ast, err := newProgram(ctx, cfg.Program, root, client, limiter, auth, patterns, cfg.XSDs, log, trace)
+	prg, ast, err := newProgram(ctx, cfg.Program, root, getEnv(cfg.AllowedEnvironment), client, limiter, auth, patterns, cfg.XSDs, log, trace)
 	if err != nil {
 		return err
 	}
@@ -991,7 +992,19 @@ var (
 	}
 )
 
-func newProgram(ctx context.Context, src, root string, client *http.Client, limiter *rate.Limiter, auth *lib.BasicAuth, patterns map[string]*regexp.Regexp, xsd map[string]string, log *logp.Logger, trace *httplog.LoggingRoundTripper) (cel.Program, *cel.Ast, error) {
+func getEnv(allowed []string) map[string]string {
+	env := make(map[string]string)
+	for _, kv := range os.Environ() {
+		k, v, ok := strings.Cut(kv, "=")
+		if !ok || !slices.Contains(allowed, k) {
+			continue
+		}
+		env[k] = v
+	}
+	return env
+}
+
+func newProgram(ctx context.Context, src, root string, vars map[string]string, client *http.Client, limiter *rate.Limiter, auth *lib.BasicAuth, patterns map[string]*regexp.Regexp, xsd map[string]string, log *logp.Logger, trace *httplog.LoggingRoundTripper) (cel.Program, *cel.Ast, error) {
 	xml, err := lib.XML(nil, xsd)
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to build xml type hints: %w", err)
@@ -1013,6 +1026,7 @@ func newProgram(ctx context.Context, src, root string, client *http.Client, limi
 		lib.Limit(limitPolicies),
 		lib.Globals(map[string]interface{}{
 			"useragent": userAgent,
+			"env":       vars,
 		}),
 	}
 	if len(patterns) != 0 {

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

@@ -365,6 +365,52 @@ var inputTests = []struct {
 			{"message": "Hello, Void!"},
 		},
 	},
+	{
+		name: "env_var_static",
+		config: map[string]interface{}{
+			"interval": 1,
+			"allowed_environment": []string{
+				"CELTESTENVVAR",
+				"NONCELTESTENVVAR",
+			},
+			"program": `{"events":[
+				{"message":env.?CELTESTENVVAR.orValue("not present")},
+				{"message":env.?NONCELTESTENVVAR.orValue("not present")},
+				{"message":env.?DISALLOWEDCELTESTENVVAR.orValue("not present")},
+			]}`,
+			"state": nil,
+			"resource": map[string]interface{}{
+				"url": "",
+			},
+		},
+		want: []map[string]interface{}{
+			{"message": "TESTVALUE"},
+			{"message": "not present"},
+			{"message": "not present"},
+		},
+	},
+	{
+		name: "env_var_dynamic",
+		config: map[string]interface{}{
+			"interval": 1,
+			"allowed_environment": []string{
+				"CELTESTENVVAR",
+				"NONCELTESTENVVAR",
+			},
+			"program": `{"events": ["CELTESTENVVAR","NONCELTESTENVVAR","DISALLOWEDCELTESTENVVAR"].map(k,
+				{"message":env[?k].orValue("not present")}
+			)}`,
+			"state": nil,
+			"resource": map[string]interface{}{
+				"url": "",
+			},
+		},
+		want: []map[string]interface{}{
+			{"message": "TESTVALUE"},
+			{"message": "not present"},
+			{"message": "not present"},
+		},
+	},
 
 	// FS-based tests.
 	{
@@ -1645,6 +1691,10 @@ func TestInput(t *testing.T) {
 		"ndjson_log_file_simple_file_scheme": "Path handling on Windows is incompatible with url.Parse/url.URL.String. See go.dev/issue/6027.",
 	}
 
+	// Set a var that is available to test env look-up.
+	os.Setenv("CELTESTENVVAR", "TESTVALUE")
+	os.Setenv("DISALLOWEDCELTESTENVVAR", "DISALLOWEDTESTVALUE")
+
 	logp.TestingSetup()
 	for _, test := range inputTests {
 		t.Run(test.name, func(t *testing.T) {