[8.19](backport #47260) Add AWS auth method for cel and httpjson inputs

changelog/fragments/1761377900-input-auth-method-aws.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: feature
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: Add AWS auth method for CEL and HTTP JSON inputs.
+
+# 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_auth.go

@@ -21,13 +21,15 @@ import (
 	"golang.org/x/oauth2/google"
 
 	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/x-pack/libbeat/common/aws"
 )
 
 type authConfig struct {
-	Basic  *basicAuthConfig  `config:"basic"`
-	Token  *tokenAuthConfig  `config:"token"`
-	Digest *digestAuthConfig `config:"digest"`
-	OAuth2 *oAuth2Config     `config:"oauth2"`
+	Basic  *basicAuthConfig       `config:"basic"`
+	Token  *tokenAuthConfig       `config:"token"`
+	Digest *digestAuthConfig      `config:"digest"`
+	OAuth2 *oAuth2Config          `config:"oauth2"`
+	AWS    *aws.SignerInputConfig `config:"aws"`
 }
 
 func (c authConfig) Validate() error {
@@ -44,6 +46,9 @@ func (c authConfig) Validate() error {
 	if c.OAuth2.isEnabled() {
 		n++
 	}
+	if c.AWS.IsEnabled() {
+		n++
+	}
 	if n > 1 {
 		return errors.New("only one kind of auth can be enabled")
 	}

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

@@ -51,6 +51,7 @@
 	"github.com/elastic/beats/v7/libbeat/version"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/internal/httplog"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/internal/httpmon"
+	"github.com/elastic/beats/v7/x-pack/libbeat/common/aws"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/mapstr"
 	"github.com/elastic/elastic-agent-libs/monitoring"
@@ -100,7 +101,7 @@
 func (input) Name() string { return inputName }
 
 func (input) Test(src inputcursor.Source, _ v2.TestContext) error {
 	cfg := src.(*source).cfg
 	if !wantClient(cfg) {
 		return nil
 	}
@@ -110,7 +111,7 @@
 // Run starts the input and blocks until it ends completes. It will return on
 // context cancellation or type invalidity errors, any other error will be retried.
 func (input) Run(env v2.Context, src inputcursor.Source, crsr inputcursor.Cursor, pub inputcursor.Publisher) error {
 	dataStreamName := src.(*source).cfg.DataStream // May be empty.
 
 	var cursor map[string]interface{}
 	env.UpdateStatus(status.Starting, dataStreamName)
@@ -213,6 +214,7 @@
 			Value: cfg.Auth.Token.Value,
 		}
 	}
+
 	wantDump := cfg.FailureDump.enabled() && cfg.FailureDump.Filename != ""
 	doCov := cfg.RecordCoverage && log.IsDebug()
 	httpOptions := lib.HTTPOptions{
@@ -864,6 +866,15 @@
 			Password:  cfg.Auth.Digest.Password,
 			NoReuse:   noReuse,
 		}
+	} else if cfg.Auth.AWS.IsEnabled() {
+		// this transport runs after the other ones (the other ones wrap this one); just to be on the safe side.
+		// If any of the other transports add any header, it must happen before the signing.
+		tr, err := aws.InitializeSignerTransport(*cfg.Auth.AWS, log, c.Transport)
+		if err != nil {
+			log.Errorw("failed to initialize aws config failed for signer", "error", err)
+			return nil, nil, err
+		}
+		c.Transport = tr
 	}
 
 	var trace *httplog.LoggingRoundTripper

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

@@ -19,6 +19,7 @@
 	"path/filepath"
 	"reflect"
 	"runtime"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -615,9 +616,9 @@
   </item>
 </order>
 `
 				io.ReadAll(r.Body)
 				r.Body.Close()
 				w.Write([]byte(text))
 			})
 			server := httptest.NewServer(r)
 			config["resource.url"] = server.URL
@@ -749,7 +750,7 @@
 				msg = fmt.Sprintf(`{"error":"expected method was %#q"}`, http.MethodGet)
 			}
 
 			w.Write([]byte(msg))
 		},
 		want: []map[string]interface{}{
 			{
@@ -830,7 +831,7 @@
 		},
 		handler: func(w http.ResponseWriter, r *http.Request) {
 			enc := json.NewEncoder(w)
 			enc.Encode(map[string][]any{"events": {r.Header.Get("foo")}})
 		},
 		want: []map[string]interface{}{
 			{
@@ -1890,6 +1891,44 @@
 		},
 	},
 
+	{
+		name: "Auth AWS V4 Signer",
+		server: func(t *testing.T, h http.HandlerFunc, config map[string]interface{}) {
+			s := httptest.NewServer(h)
+			config["resource.url"] = s.URL
+			t.Cleanup(s.Close)
+		},
+		config: map[string]interface{}{
+			"interval":                   1,
+			"auth.aws.access_key_id":     "AKIAIOSFODNN7EXAMPLE",
+			"auth.aws.secret_access_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+			"auth.aws.default_region":    "us-east-1",
+			"auth.aws.service_name":      "guardduty",
+			"program": `
+	bytes(get(state.url).Body).as(body, {
+		"events": [body.decode_json()]
+	})
+	`,
+		},
+		handler: awsAuthHandler("AKIAIOSFODNN7EXAMPLE", defaultHandler(http.MethodGet, "")),
+		want: []map[string]interface{}{
+			{
+				"hello": []interface{}{
+					map[string]interface{}{
+						"world": "moon",
+					},
+					map[string]interface{}{
+						"space": []interface{}{
+							map[string]interface{}{
+								"cake": "pumpkin",
+							},
+						},
+					},
+				},
+			},
+		},
+	},
+
 	// Multi-step requests.
 	{
 		name:   "simple_multistep_GET_request",
@@ -2495,6 +2534,24 @@
 	}
 }
 
+func awsAuthHandler(expectedTokenID string, handle http.HandlerFunc) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		authHeader := r.Header.Get("Authorization")
+		if !strings.HasPrefix(authHeader, fmt.Sprintf("AWS4-HMAC-SHA256 Credential=%s/", expectedTokenID)) {
+			http.Error(w, `{"error":"not authorized"}`, http.StatusBadRequest)
+			return
+		}
+
+		amzDate := r.Header.Get("X-Amz-Date")
+		if amzDate == "" {
+			http.Error(w, `{"error":"not authorized"}`, http.StatusBadRequest)
+			return
+		}
+
+		handle(w, r)
+	}
+}
+
 //nolint:errcheck // No point checking errors in test server.
 func digestAuthHandler(user, pass, realm, nonce string, handle http.HandlerFunc) http.HandlerFunc {
 	chal := &digest.Challenge{

x-pack/filebeat/input/httpjson/config_auth.go

@@ -20,15 +20,27 @@ import (
 	"golang.org/x/oauth2/google"
 
 	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/x-pack/libbeat/common/aws"
 )
 
 type authConfig struct {
-	Basic  *basicAuthConfig `config:"basic"`
-	OAuth2 *oAuth2Config    `config:"oauth2"`
+	Basic  *basicAuthConfig       `config:"basic"`
+	OAuth2 *oAuth2Config          `config:"oauth2"`
+	AWS    *aws.SignerInputConfig `config:"aws"`
 }
 
 func (c authConfig) Validate() error {
-	if c.Basic.isEnabled() && c.OAuth2.isEnabled() {
+	var n int
+	if c.Basic.isEnabled() {
+		n++
+	}
+	if c.OAuth2.isEnabled() {
+		n++
+	}
+	if c.AWS.IsEnabled() {
+		n++
+	}
+	if n > 1 {
 		return errors.New("only one kind of auth can be enabled")
 	}
 	return nil

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

@@ -38,6 +38,7 @@
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/internal/httplog"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/internal/httpmon"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/internal/private"
+	"github.com/elastic/beats/v7/x-pack/libbeat/common/aws"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/mapstr"
 	"github.com/elastic/elastic-agent-libs/monitoring"
@@ -105,7 +106,7 @@
 func (r redact) MarshalLogObject(enc zapcore.ObjectEncoder) error {
 	v, err := private.Redact(r.value, "", r.fields)
 	if err != nil {
 		return fmt.Errorf("could not redact value: %v", err)
 	}
 	return v.MarshalLogObject(enc)
 }
@@ -304,7 +305,22 @@
 		client *http.Client
 		err    error
 	)
-	if authCfg.OAuth2.isEnabled() {
+	switch {
+	case authCfg.AWS.IsEnabled():
+		client, err = newNetHTTPClient(ctx, requestCfg, log, reg)
+		if err != nil {
+			log.Errorw("creation of initial http client failed", "error", err)
+			return nil, err
+		}
+
+		log.Debugw("creating signer", "region", authCfg.AWS.DefaultRegion, "service", authCfg.AWS.ServiceName)
+		tr, err := aws.InitializeSignerTransport(*authCfg.AWS, log, client.Transport)
+		if err != nil {
+			log.Errorw("failed to initialize aws config failed for signer", "error", err)
+			return nil, err
+		}
+		client.Transport = tr
+	case authCfg.OAuth2.isEnabled():
 		client = authCfg.OAuth2.prepared
 		if client == nil {
 			client, err = newNetHTTPClient(ctx, requestCfg, log, reg)
@@ -317,7 +333,7 @@
 			}
 			authCfg.OAuth2.prepared = client
 		}
-	} else {
+	default:
 		client, err = newNetHTTPClient(ctx, requestCfg, log, reg)
 		if err != nil {
 			return nil, err

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

@@ -12,6 +12,7 @@
 	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
@@ -653,6 +654,24 @@
 		handler:  oauth2Handler,
 		expected: []string{`{"hello": "world"}`},
 	},
+	{
+		name: "aws auth",
+		setupServer: func(t testing.TB, h http.HandlerFunc, config map[string]interface{}) {
+			server := httptest.NewServer(h)
+			config["request.url"] = server.URL
+			t.Cleanup(server.Close)
+		},
+		baseConfig: map[string]interface{}{
+			"interval":                   1,
+			"request.method":             http.MethodGet,
+			"auth.aws.access_key_id":     "AKIAIOSFODNN7EXAMPLE",
+			"auth.aws.secret_access_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+			"auth.aws.default_region":    "us-east-1",
+			"auth.aws.service_name":      "guardduty",
+		},
+		handler:  awsAuthHandler("AKIAIOSFODNN7EXAMPLE", defaultHandler(http.MethodGet, "", "")),
+		expected: []string{`{"hello":[{"world":"moon"},{"space":[{"cake":"pumpkin"}]}]}`},
+	},
 	{
 		name: "request_transforms_can_access_state_from_previous_transforms",
 		setupServer: func(t testing.TB, h http.HandlerFunc, config map[string]interface{}) {
@@ -1528,9 +1547,9 @@
   </item>
 </order>
 `
 				io.ReadAll(r.Body)
 				r.Body.Close()
 				w.Write([]byte(text))
 			})
 			server := httptest.NewServer(r)
 			config["request.url"] = server.URL
@@ -1683,7 +1702,7 @@
 				case got := <-chanClient.Channel:
 					val, err := got.Fields.GetValue("message")
 					assert.NoError(t, err)
 					assert.JSONEq(t, test.expected[receivedCount], val.(string))
 					receivedCount += 1
 					if receivedCount == len(test.expected) {
 						cancel()
@@ -1889,6 +1908,24 @@
 	}
 }
 
+func awsAuthHandler(expectedTokenID string, handle http.HandlerFunc) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		authHeader := r.Header.Get("Authorization")
+		if !strings.HasPrefix(authHeader, fmt.Sprintf("AWS4-HMAC-SHA256 Credential=%s/", expectedTokenID)) {
+			http.Error(w, `{"error":"not authorized"}`, http.StatusBadRequest)
+			return
+		}
+
+		amzDate := r.Header.Get("X-Amz-Date")
+		if amzDate == "" {
+			http.Error(w, `{"error":"not authorized"}`, http.StatusBadRequest)
+			return
+		}
+
+		handle(w, r)
+	}
+}
+
 func rateLimitHandler() http.HandlerFunc {
 	var isRetry bool
 	return func(w http.ResponseWriter, r *http.Request) {