[libbeat] Cloud Connectors AWS implementation (#47587)

moukoublen · web-flow · commit 78d71d3ab89a · 2025-11-26T16:16:49.000+01:00
As discussed [here](elastic/security-team#13112), this PR introduces the cloud connectors flow into libbeat to enable its use in agentless AWS integrations (in combination with the new aws [auth method](#47260)). Cloud Connectors flow in high level: <img width="1391" height="563" alt="Screenshot 2025-11-12 at 10 49 01 AM" src="https://github.com/user-attachments/assets/2d0855da-e259-47d4-b294-3f755f078c23" /> * User creates (through a cloud formation provided by Elastic) a role that has all the necessary permissions required by the integration functionality that trusts Elastic Global Role to be assumed by. (The cloud formation also creates a unique ID as an external ID that the user adds to the package policy.) * Elastic Global Role trusts/federates with the agnentless OIDC Issuer ([doc](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_providers_create_oidc.html)), which means id token created from that issuer can assume the Elastic Global Role. * Agentless OIDC Issuer provides each agentless pod with an ID token. * User creates an agentless agent and provides the Remote Role ARN and the External ID to the package policy (external id is treated as secret). * Beat performs a [role chaining](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles.html#iam-term-role-chaining) to the Remote Role through the Elastic Global Role (described below). * During the last role-assume (of the user's remote role) the [source identity](https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRole.html#:~:text=Required%3A%20No-,SourceIdentity,-The%20source%20identity) is set to the elastic cloud resource id. This should be expected in the remote role trust policy as a condition, and it ensures that, in order for the role-assume to be done, both user input (role arn, external id) and system-provided information (resource id) are correct. Example remote role trust policy condition: ```json "Condition": { "StringEquals": { "sts:ExternalId": "randomly-generated-in-customer-csp-treated-as-secret", "sts:SourceIdentity": "elastic resouce id (like project id in serverless)" } } ``` ## Proposed commit message A new boolean flag was introduced in the ConfigAWS module (`x-pack/libbeat/common/aws/credentials.go`) to indicate whether cloud connectors are supported. When this setting is set to `true`, the initialization of the aws credentials follows this chain: 1. Assume the Elastic Global Role [with web identity](https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRoleWithWebIdentity.html) using the ID token that the oidc issuer has provided to the pod. The Elastic Global Role ARN and the ID token file path, are set by agentless controller as an env vars in agentless agents pods. 2. [Assumes](https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRole.html) the Remote Role using the role arn and the external id provided by the package policy (and thus the user).
diff --git a/changelog/fragments/1762938116-aws-cloud-connectors.yaml b/changelog/fragments/1762938116-aws-cloud-connectors.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: Introduce cloud connectors flow.
+
+# 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: "all"
+
+# 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
diff --git a/x-pack/libbeat/common/aws/cloud_connectors.go b/x-pack/libbeat/common/aws/cloud_connectors.go
@@ -0,0 +1,111 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package aws
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"time"
+
+	awssdk "github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/credentials/stscreds"
+	"github.com/aws/aws-sdk-go-v2/service/sts"
+
+	"github.com/elastic/elastic-agent-libs/logp"
+)
+
+// These env vars are provided by agentless controller when the cloud connectors flow is enabled.
+const (
+	CloudConnectorsGlobalRoleEnvVar      = "CLOUD_CONNECTORS_GLOBAL_ROLE"
+	CloudConnectorsJWTPathEnvVar         = "CLOUD_CONNECTORS_ID_TOKEN_FILE"
+	CloudConnectorsCloudResourceIDEnvVar = "CLOUD_RESOURCE_ID"
+)
+
+// CloudConnectorsConfig is the config for the cloud connectors flow
+type CloudConnectorsConfig struct {
+	ElasticGlobalRoleARN string
+	IDTokenPath          string
+	CloudResourceID      string
+}
+
+func parseCloudConnectorsConfigFromEnv() (CloudConnectorsConfig, error) {
+	cc := CloudConnectorsConfig{
+		ElasticGlobalRoleARN: os.Getenv(CloudConnectorsGlobalRoleEnvVar),
+		IDTokenPath:          os.Getenv(CloudConnectorsJWTPathEnvVar),
+		CloudResourceID:      os.Getenv(CloudConnectorsCloudResourceIDEnvVar),
+	}
+
+	var errs []error
+
+	if cc.ElasticGlobalRoleARN == "" {
+		errs = append(errs, errors.New("elastic global role arn is not configured"))
+	}
+	if cc.IDTokenPath == "" {
+		errs = append(errs, errors.New("id token path is not configured"))
+	}
+	if cc.CloudResourceID == "" {
+		errs = append(errs, errors.New("cloud resource id is not configured"))
+	}
+
+	if len(errs) > 0 {
+		return CloudConnectorsConfig{}, fmt.Errorf("cloud connectors config is invalid: %w", errors.Join(errs...))
+	}
+
+	return cc, nil
+}
+
+const defaultIntermediateDuration = 20 * time.Minute
+
+func addCloudConnectorsCredentials(config ConfigAWS, cloudConnectorsConfig CloudConnectorsConfig, awsConfig *awssdk.Config, logger *logp.Logger) {
+	logger = logger.Named("addCloudConnectorsCredentials")
+	logger.Debug("Switching credentials provider to Cloud Connectors")
+
+	addCredentialsChain(
+		awsConfig,
+
+		// Step 1: Assume the Elastic Global Role with web identity using the ID token provided by the agentless OIDC issuer.
+		func(c awssdk.Config) awssdk.CredentialsProvider {
+			provider := stscreds.NewWebIdentityRoleProvider(
+				sts.NewFromConfig(c), // client uses credentials from previous config.
+				cloudConnectorsConfig.ElasticGlobalRoleARN,
+				stscreds.IdentityTokenFile(cloudConnectorsConfig.IDTokenPath),
+				func(opt *stscreds.WebIdentityRoleOptions) {
+					opt.Duration = defaultIntermediateDuration
+				},
+			)
+			return awssdk.NewCredentialsCache(provider)
+		},
+
+		// Step 2: Assume the remote role (the user's configured role), using the previously assumed role in the chain.
+		func(c awssdk.Config) awssdk.CredentialsProvider {
+			assumeRoleProvider := stscreds.NewAssumeRoleProvider(
+				sts.NewFromConfig(c), // client uses credentials from previous config.
+				config.RoleArn,
+				func(aro *stscreds.AssumeRoleOptions) {
+					aro.Duration = config.AssumeRoleDuration
+					if config.ExternalID != "" {
+						aro.ExternalID = awssdk.String(config.ExternalID)
+
+						// The source identity is set by the system (env var) rather than user input (package policy).
+						// It should be required by the remote role (the role to assume) as a condition for assuming it.
+						aro.SourceIdentity = awssdk.String(cloudConnectorsConfig.CloudResourceID)
+					}
+				},
+			)
+			return awssdk.NewCredentialsCache(assumeRoleProvider, func(options *awssdk.CredentialsCacheOptions) {
+				if config.AssumeRoleExpiryWindow > 0 {
+					options.ExpiryWindow = config.AssumeRoleExpiryWindow
+				}
+			})
+		},
+	)
+}
+
+func addCredentialsChain(awsConfig *awssdk.Config, chain ...func(awssdk.Config) awssdk.CredentialsProvider) {
+	for _, fn := range chain {
+		awsConfig.Credentials = fn(*awsConfig)
+	}
+}
diff --git a/x-pack/libbeat/common/aws/cloud_connectors_test.go b/x-pack/libbeat/common/aws/cloud_connectors_test.go
@@ -0,0 +1,177 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package aws
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/url"
+	"os"
+	"path"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/sts"
+	"github.com/aws/aws-sdk-go-v2/service/sts/types"
+	"github.com/aws/smithy-go/middleware"
+	smithyhttp "github.com/aws/smithy-go/transport/http"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/elastic-agent-libs/logp/logptest"
+)
+
+func TestAddCloudConnectorsCredentials(t *testing.T) {
+	config := ConfigAWS{
+		RoleArn:                "arn:aws:iam::123456789012:role/customer-role",
+		ExternalID:             "external-id-456",
+		AssumeRoleDuration:     2 * time.Hour,
+		AssumeRoleExpiryWindow: 10 * time.Minute,
+	}
+	cloudConnectorsConfig := CloudConnectorsConfig{
+		ElasticGlobalRoleARN: "arn:aws:iam::999999999999:role/elastic-global-role",
+		CloudResourceID:      "abcd1234",
+	}
+	tokenFileContent := "abc123"
+
+	tmpDir := t.TempDir()
+	pth := path.Join(tmpDir, "id_token")
+	_ = os.WriteFile(path.Join(tmpDir, "id_token"), []byte(tokenFileContent), 0o644)
+	cloudConnectorsConfig.IDTokenPath = pth
+
+	// Create a base AWS config
+	awsConfig := &aws.Config{
+		Region:       "us-east-1",
+		BaseEndpoint: aws.String("https://aws.mock"),
+	}
+
+	// Create a test logger
+	logger := logptest.NewTestingLogger(t, "")
+
+	// mock responses
+	receivedCalls := 0
+	awsConfig.APIOptions = append(awsConfig.APIOptions, func(stack *middleware.Stack) error {
+		return stack.Finalize.Add(
+			middleware.FinalizeMiddlewareFunc(
+				"mock",
+				func(ctx context.Context, in middleware.FinalizeInput, next middleware.FinalizeHandler) (middleware.FinalizeOutput, middleware.Metadata, error) {
+					req, is := in.Request.(*smithyhttp.Request)
+					require.Truef(t, is, "request expected to be of type *smithyhttp.Request, got: %T", in.Request)
+					receivedCalls++
+					bd, err := io.ReadAll(req.GetStream())
+					assert.NoError(t, req.RewindStream())
+					assert.NoError(t, err)
+					body := string(bd)
+
+					switch receivedCalls {
+
+					// Expect the first request to be AssumeRoleWithWebIdentity
+					case 1:
+						q, err := url.ParseQuery(body)
+						assert.NoError(t, err)
+						assert.Equal(t, "AssumeRoleWithWebIdentity", q.Get("Action"))
+						assert.Equal(t, "1200", q.Get("DurationSeconds"))
+						assert.Equal(t, cloudConnectorsConfig.ElasticGlobalRoleARN, q.Get("RoleArn"))
+						assert.Equal(t, tokenFileContent, q.Get("WebIdentityToken"))
+						return middleware.FinalizeOutput{
+							Result: &sts.AssumeRoleWithWebIdentityOutput{
+								Credentials: &types.Credentials{
+									AccessKeyId:     aws.String("AKIAFAKEEXAMPLE00001"),
+									SecretAccessKey: aws.String("FAKEwJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY1"),
+									SessionToken:    aws.String("FwoGZXIvYXdzEFAaDFAKESESSIONTOKENEXAMPLE1"),
+									Expiration:      aws.Time(time.Now().Add(defaultIntermediateDuration)),
+								},
+							},
+						}, middleware.Metadata{}, nil
+
+					// Expect the second request to be AssumeRole
+					case 2:
+						q, err := url.ParseQuery(body)
+						assert.NoError(t, err)
+						assert.Equal(t, "AssumeRole", q.Get("Action"))
+						assert.Equal(t, "7200", q.Get("DurationSeconds"))
+						assert.Equal(t, config.ExternalID, q.Get("ExternalId"))
+						assert.Equal(t, config.RoleArn, q.Get("RoleArn"))
+						assert.Equal(t, "abcd1234", q.Get("SourceIdentity"))
+						return middleware.FinalizeOutput{
+							Result: &sts.AssumeRoleOutput{
+								Credentials: &types.Credentials{
+									AccessKeyId:     aws.String("AKIAFAKEEXAMPLE00002"),
+									SecretAccessKey: aws.String("FAKEwJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY2"),
+									SessionToken:    aws.String("FwoGZXIvYXdzEFAaDFAKESESSIONTOKENEXAMPLE2"),
+									Expiration:      aws.Time(time.Now().Add(defaultIntermediateDuration)),
+								},
+							},
+						}, middleware.Metadata{}, nil
+
+					default:
+						t.Fatal("unexpected aws sdk call")
+						return middleware.FinalizeOutput{}, middleware.Metadata{}, fmt.Errorf("unexpected operation")
+					}
+				},
+			),
+			middleware.After,
+		)
+	})
+
+	// Call the function under test
+	addCloudConnectorsCredentials(
+		config,
+		cloudConnectorsConfig,
+		awsConfig,
+		logger,
+	)
+
+	// Verify that credentials provider was set
+	require.NotNil(t, awsConfig.Credentials, "credentials provider should be set")
+
+	crd, err := awsConfig.Credentials.Retrieve(t.Context())
+	require.NoError(t, err)
+	require.NotNil(t, crd)
+	require.Equal(t, 2, receivedCalls)
+}
+
+func TestParseCloudConnectorsConfigFromEnv(t *testing.T) {
+	t.Run("happy_path", func(t *testing.T) {
+		t.Setenv(CloudConnectorsGlobalRoleEnvVar, "arn:aws:iam::999999999999:role/elastic-global-role")
+		t.Setenv(CloudConnectorsJWTPathEnvVar, "/path/token")
+		t.Setenv(CloudConnectorsCloudResourceIDEnvVar, "abc123")
+
+		got, err := parseCloudConnectorsConfigFromEnv()
+
+		require.NoError(t, err)
+
+		assert.Equal(
+			t,
+			CloudConnectorsConfig{
+				ElasticGlobalRoleARN: "arn:aws:iam::999999999999:role/elastic-global-role",
+				IDTokenPath:          "/path/token",
+				CloudResourceID:      "abc123",
+			},
+			got,
+		)
+	})
+
+	t.Run("missing config single", func(t *testing.T) {
+		t.Setenv(CloudConnectorsGlobalRoleEnvVar, "arn:aws:iam::999999999999:role/elastic-global-role")
+		t.Setenv(CloudConnectorsJWTPathEnvVar, "/path/token")
+
+		got, err := parseCloudConnectorsConfigFromEnv()
+
+		require.ErrorContains(t, err, "cloud resource id")
+		assert.Equal(t, CloudConnectorsConfig{}, got)
+	})
+
+	t.Run("missing config all", func(t *testing.T) {
+		got, err := parseCloudConnectorsConfigFromEnv()
+
+		require.ErrorContains(t, err, "elastic global role")
+		require.ErrorContains(t, err, "id token")
+		require.ErrorContains(t, err, "cloud resource id")
+		assert.Equal(t, CloudConnectorsConfig{}, got)
+	})
+}
diff --git a/x-pack/libbeat/common/aws/credentials.go b/x-pack/libbeat/common/aws/credentials.go
@@ -52,6 +52,10 @@ type ConfigAWS struct {
 	// AssumeRoleExpiryWindow will allow the credentials to trigger refreshing prior to the credentials
 	// actually expiring. If expiry_window is less than or equal to zero, the setting is ignored.
 	AssumeRoleExpiryWindow time.Duration `config:"assume_role.expiry_window"`
+
+	// UseCloudConnectors indicates whether the cloud connectors flow is used.
+	// If this is true, the InitializeAWSConfig should initialize the AWS cloud connector role chaining flow.
+	UseCloudConnectors bool `config:"use_cloud_connectors"`
 }
 
 // InitializeAWSConfig function creates the awssdk.Config object from the provided config
@@ -66,10 +70,19 @@ func InitializeAWSConfig(beatsConfig ConfigAWS, logger *logp.Logger) (awssdk.Con
 	}
 
 	// Assume IAM role if iam_role config parameter is given
-	if beatsConfig.RoleArn != "" {
+	if beatsConfig.RoleArn != "" && !beatsConfig.UseCloudConnectors {
 		addAssumeRoleProviderToAwsConfig(beatsConfig, &awsConfig, logger)
 	}
 
+	// If cloud connectors method is selected from config, initialize the role chaining.
+	if beatsConfig.UseCloudConnectors {
+		cloudConnectorsConfig, err := parseCloudConnectorsConfigFromEnv()
+		if err != nil {
+			return awsConfig, err
+		}
+		addCloudConnectorsCredentials(beatsConfig, cloudConnectorsConfig, &awsConfig, logger)
+	}
+
 	var proxy func(*http.Request) (*url.URL, error)
 	if beatsConfig.ProxyUrl != "" {
 		proxyUrl, err := httpcommon.NewProxyURIFromString(beatsConfig.ProxyUrl)
@@ -142,7 +155,11 @@ func getConfigSharedCredentialProfile(beatsConfig ConfigAWS, logger *logp.Logger
 		return cfg, fmt.Errorf("awsConfig.LoadDefaultConfig failed with shared credential profile given: [%w]", err)
 	}
 
-	logger.Debug("Using shared credential profile for AWS credential")
+	if beatsConfig.ProfileName != "" || beatsConfig.SharedCredentialFile != "" {
+		logger.Debug("Using shared credential profile for AWS credential")
+	} else {
+		logger.Debug("Using default config for AWS")
+	}
 	return cfg, nil
 }
 
diff --git a/x-pack/libbeat/common/aws/credentials_test.go b/x-pack/libbeat/common/aws/credentials_test.go
