x-pack/filebeat/input/streaming: add configurable User-Agent header

Add a user_agent config field and a transport-level decorator that
sets the User-Agent header on all outbound HTTP requests made by
the CrowdStrike streaming input (discover, firehose, session
refresh, and OAuth2 token fetch). The decorator unconditionally
overwrites Go's default "Go-http-client/1.1" which http.Client.Do
sets before calling RoundTrip.

When user_agent is empty, the Elastic Agent's built-in user agent
string is used as the default.

This is needed for CrowdStrike technology partner compliance, which
requires a User-Agent identifying the integration and version on
every API request.

Author: efd6

changelog/fragments/1783565292-streaming-user-agent.yaml

@@ -0,0 +1,3 @@
+kind: enhancement
+summary: Add configurable User-Agent header to the CrowdStrike streaming input.
+component: filebeat

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

@@ -63,6 +63,11 @@ type config struct {
 	// domain as the configured URL. Entries here extend the set
 	// of accepted origins.
 	ResourceOrigins []string `config:"resource_origins"`
+	// UserAgent overrides the default User-Agent header sent on
+	// all outbound HTTP requests (discover, firehose, session
+	// refresh, and OAuth2 token fetch). When empty, the Elastic
+	// Agent's built-in user agent string is used.
+	UserAgent string `config:"user_agent"`
 }
 
 type redact struct {

x-pack/filebeat/input/streaming/crowdstrike.go

@@ -197,6 +197,11 @@ func NewFalconHoseFollower(ctx context.Context, env v2.Context, cfg config, curs
 		s.allowedOrigins = append(s.allowedOrigins, o)
 	}
 
+	ua := cfg.UserAgent
+	if ua == "" {
+		ua = env.Agent.UserAgent
+	}
+
 	// Build the auth transport before zeroing timeouts for the streaming
 	// client. The oauth2 token endpoint needs normal request timeouts;
 	// moving this after the timeout zeroing will cause auth requests to
@@ -210,7 +215,7 @@ func NewFalconHoseFollower(ctx context.Context, env v2.Context, cfg config, curs
 		now = time.Now
 	}
 	s.authTransport = &rateLimitTransport{
-		base:     authClient.Transport,
+		base:     userAgentTransport{ua: ua, base: authClient.Transport},
 		timeout:  authClient.Timeout,
 		maxRetry: 3,
 		wait:     60 * time.Second,
@@ -225,10 +230,27 @@ func NewFalconHoseFollower(ctx context.Context, env v2.Context, cfg config, curs
 		stat.UpdateStatus(status.Failed, "failed to configure client: "+err.Error())
 		return nil, err
 	}
+	s.plainClient.Transport = userAgentTransport{ua: ua, base: s.plainClient.Transport}
 
 	return &s, nil
 }
 
+var _ http.RoundTripper = userAgentTransport{}
+
+// userAgentTransport wraps an http.RoundTripper and unconditionally
+// sets the User-Agent header on every outgoing request. It overwrites
+// Go's default "Go-http-client/1.1" which http.Client.Do sets before
+// calling RoundTrip.
+type userAgentTransport struct {
+	ua   string
+	base http.RoundTripper
+}
+
+func (t userAgentTransport) RoundTrip(r *http.Request) (*http.Response, error) {
+	r.Header.Set("User-Agent", t.ua)
+	return t.base.RoundTrip(r)
+}
+
 // FollowStream receives, processes and publishes events from the subscribed
 // FalconHose stream.
 func (s *falconHoseStream) FollowStream(ctx context.Context) error {

x-pack/filebeat/input/streaming/crowdstrike_unit_test.go

@@ -12,20 +12,18 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
-	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/logp/logptest"
 	"github.com/elastic/elastic-agent-libs/monitoring"
 
 	cursor "github.com/elastic/beats/v7/filebeat/input/v2/input-cursor"
 )
 
 func TestFollowSession_FirehoseHTTPError(t *testing.T) {
-	logp.TestingSetup()
-
 	tests := []struct {
 		name       string
 		statusCode int
@@ -118,8 +116,6 @@ func TestFollowSession_DiscoverGETFailureIsTransient(t *testing.T) {
 }
 
 func TestFollowSession_NonObjectMessage(t *testing.T) {
-	logp.TestingSetup()
-
 	validEvent := `{"metadata":{"eventType":"TestEvent","offset":1},"event":{"TestField":"value"}}`
 
 	tests := []struct {
@@ -176,6 +172,129 @@ func TestFollowSession_NonObjectMessage(t *testing.T) {
 	}
 }
 
+func TestUserAgentTransport(t *testing.T) {
+	const want = "Elastic-crowdstrike/4.0.0"
+
+	var got string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		got = r.Header.Get("User-Agent")
+	}))
+	defer srv.Close()
+
+	cli := srv.Client()
+	cli.Transport = userAgentTransport{ua: want, base: cli.Transport}
+
+	// http.Client.Do sets "Go-http-client/1.1" before RoundTrip;
+	// the transport must overwrite it.
+	req, err := http.NewRequest("GET", srv.URL, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp, err := cli.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp.Body.Close()
+
+	if got != want {
+		t.Errorf("User-Agent = %q; want %q", got, want)
+	}
+}
+
+func TestFollowSession_UserAgent(t *testing.T) {
+	const want = "Elastic-crowdstrike/4.0.0"
+
+	type requestRecord struct {
+		path      string
+		userAgent string
+	}
+	var (
+		mu       sync.Mutex
+		requests []requestRecord
+	)
+	record := func(r *http.Request) {
+		mu.Lock()
+		requests = append(requests, requestRecord{path: r.URL.Path, userAgent: r.Header.Get("User-Agent")})
+		mu.Unlock()
+	}
+
+	firehoseSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		record(r)
+		switch {
+		case strings.HasPrefix(r.URL.Path, "/firehose"):
+			// Send one event then EOF to end the session cleanly.
+			fmt.Fprintln(w, `{"metadata":{"eventType":"Test","offset":1},"event":{"field":"value"}}`)
+		case strings.HasPrefix(r.URL.Path, "/refresh"):
+			w.WriteHeader(http.StatusOK)
+		}
+	}))
+	defer firehoseSrv.Close()
+
+	discoverSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		record(r)
+		w.Header().Set("Content-Type", "application/json")
+		// refreshActiveSessionInterval is short but refreshSessionWait
+		// clamps to a 15s floor, so no refresh fires during the test.
+		resp := map[string]any{
+			"resources": []map[string]any{
+				{
+					"dataFeedURL": firehoseSrv.URL + "/firehose",
+					"sessionToken": map[string]any{
+						"token":      "test-token",
+						"expiration": "2099-01-01T00:00:00Z",
+					},
+					"refreshActiveSessionURL":      firehoseSrv.URL + "/refresh",
+					"refreshActiveSessionInterval": 1,
+				},
+			},
+			"meta": map[string]any{},
+		}
+		b, _ := json.Marshal(resp)
+		w.Write(b)
+	}))
+	defer discoverSrv.Close()
+
+	// Build clients wrapped with the userAgentTransport, matching the
+	// production wiring in NewFalconHoseFollower.
+	authClient := discoverSrv.Client()
+	authClient.Transport = userAgentTransport{ua: want, base: authClient.Transport}
+	plainClient := firehoseSrv.Client()
+	plainClient.Transport = userAgentTransport{ua: want, base: plainClient.Transport}
+
+	pub := new(countingPublisher)
+	s := newTestStreamWithPublisher(t, discoverSrv.URL, plainClient, pub)
+
+	state := map[string]any{}
+	_, err := s.followSession(context.Background(), authClient, state)
+	if err != nil {
+		t.Fatalf("followSession() unexpected error: %v", err)
+	}
+
+	// Check that the discover and firehose requests carried the custom
+	// User-Agent. We don't assert on refresh because the goroutine may
+	// or may not fire within the test window.
+	mu.Lock()
+	defer mu.Unlock()
+	var sawDiscover, sawFirehose bool
+	for _, r := range requests {
+		if r.userAgent != want {
+			t.Errorf("request to %s had User-Agent = %q; want %q", r.path, r.userAgent, want)
+		}
+		switch {
+		case r.path == "/" || r.path == "":
+			sawDiscover = true
+		case strings.HasPrefix(r.path, "/firehose"):
+			sawFirehose = true
+		}
+	}
+	if !sawDiscover {
+		t.Error("no discover request recorded")
+	}
+	if !sawFirehose {
+		t.Error("no firehose request recorded")
+	}
+}
+
 func discoverResponse(t *testing.T, feedURL, refreshURL string) string {
 	t.Helper()
 	resp := map[string]any{