diff --git a/internal/config/config.go b/internal/config/config.go
index c59bc3aa86..0c24ee235d 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -67,8 +67,10 @@ type CloudConfig struct {
 }
 
 type AwsConfig struct {
-	Cred        aws.ConfigAWS `config:"credentials"`
-	AccountType string        `config:"account_type"`
+	Cred                  aws.ConfigAWS `config:"credentials"`
+	AccountType           string        `config:"account_type"`
+	CloudConnectors       bool          `config:"supports_cloud_connectors"`
+	CloudConnectorsConfig CloudConnectorsConfig
 }
 
 type GcpConfig struct {
@@ -169,6 +171,10 @@ func New(cfg *config.C) (*Config, error) {
 		))
 	}
 
+	if c.CloudConfig.Aws.CloudConnectors {
+		c.CloudConfig.Aws.CloudConnectorsConfig = newCloudConnectorsConfig()
+	}
+
 	return c, nil
 }
 
@@ -203,3 +209,26 @@ func isSupportedBenchmark(benchmark string) bool {
 	}
 	return false
 }
+
+// Cloud Connectors roles and resource id must be provided by the system (controller)
+// and not user input (package policy) for security reasons.
+
+const (
+	CloudConnectorsLocalRoleEnvVar  = "CLOUD_CONNECTORS_LOCAL_ROLE"
+	CloudConnectorsGlobalRoleEnvVar = "CLOUD_CONNECTORS_GLOBAL_ROLE"
+	CloudResourceIDEnvVar           = "CLOUD_RESOURCE_ID"
+)
+
+type CloudConnectorsConfig struct {
+	LocalRoleARN  string
+	GlobalRoleARN string
+	ResourceID    string
+}
+
+func newCloudConnectorsConfig() CloudConnectorsConfig {
+	return CloudConnectorsConfig{
+		LocalRoleARN:  os.Getenv(CloudConnectorsLocalRoleEnvVar),
+		GlobalRoleARN: os.Getenv(CloudConnectorsGlobalRoleEnvVar),
+		ResourceID:    os.Getenv(CloudResourceIDEnvVar),
+	}
+}
diff --git a/internal/config/config_test.go b/internal/config/config_test.go
index b72accba9e..b980357a42 100644
--- a/internal/config/config_test.go
+++ b/internal/config/config_test.go
@@ -47,25 +47,25 @@ func (s *ConfigTestSuite) TestNew() {
 		expectedCloudConfig CloudConfig
 	}{
 		{
-			`
+			config: `
 config:
   v1:
     benchmark: cis_k8s
 `,
-			"cis_k8s",
-			CloudConfig{},
+			expectedType:        "cis_k8s",
+			expectedCloudConfig: CloudConfig{},
 		},
 		{
-			`
+			config: `
 config:
   v1:
     benchmark: cis_azure
 `,
-			"cis_azure",
-			CloudConfig{},
+			expectedType:        "cis_azure",
+			expectedCloudConfig: CloudConfig{},
 		},
 		{
-			`
+			config: `
 config:
   v1:
     benchmark: cis_eks
@@ -79,8 +79,8 @@ config:
         credential_profile_name: credential_profile_name
         role_arn: role_arn
 `,
-			"cis_eks",
-			CloudConfig{
+			expectedType: "cis_eks",
+			expectedCloudConfig: CloudConfig{
 				Aws: AwsConfig{
 					Cred: aws.ConfigAWS{
 						AccessKeyID:          "key",
@@ -229,3 +229,109 @@ revision: 1`,
 		})
 	}
 }
+
+func (s *ConfigTestSuite) TestCloudConnectorsConfig() {
+	tests := map[string]struct {
+		config              string
+		overwriteEnv        func(t *testing.T)
+		expectedType        string
+		expectedCloudConfig CloudConfig
+	}{
+		"happy path cloud connectors enabled": {
+			config: `
+config:
+  v1:
+    benchmark: cis_aws
+    aws:
+        supports_cloud_connectors: true
+        credentials:
+            external_id: abc123
+`,
+			expectedType: "cis_aws",
+			expectedCloudConfig: CloudConfig{
+				Aws: AwsConfig{
+					CloudConnectors: true,
+					Cred: aws.ConfigAWS{
+						ExternalID: "abc123",
+					},
+					CloudConnectorsConfig: CloudConnectorsConfig{},
+				},
+			},
+		},
+		"happy path cloud connectors enabled - attempt overwrite roles": {
+			config: `
+config:
+  v1:
+    benchmark: cis_aws
+    aws:
+        account_type: single-account
+        supports_cloud_connectors: true
+        credentials:
+            external_id: abc123
+        CloudConnectorsConfig:
+            LocalRoleARN: "abc123"
+            LocalRoleARN: "abc123"
+`,
+			expectedType: "cis_aws",
+			expectedCloudConfig: CloudConfig{
+				Aws: AwsConfig{
+					AccountType:     SingleAccount,
+					CloudConnectors: true,
+					Cred: aws.ConfigAWS{
+						ExternalID: "abc123",
+					},
+					CloudConnectorsConfig: CloudConnectorsConfig{},
+				},
+			},
+		},
+		"happy path cloud connectors enabled - env vars set": {
+			config: `
+config:
+  v1:
+    benchmark: cis_aws
+    aws:
+        account_type: single-account
+        supports_cloud_connectors: true
+        credentials:
+            external_id: abc123
+`,
+			overwriteEnv: func(t *testing.T) {
+				t.Helper()
+				t.Setenv(CloudConnectorsLocalRoleEnvVar, "abc123")
+				t.Setenv(CloudConnectorsGlobalRoleEnvVar, "abc456")
+				t.Setenv(CloudResourceIDEnvVar, "abc789")
+			},
+			expectedType: "cis_aws",
+			expectedCloudConfig: CloudConfig{
+				Aws: AwsConfig{
+					AccountType:     SingleAccount,
+					CloudConnectors: true,
+					Cred: aws.ConfigAWS{
+						ExternalID: "abc123",
+					},
+					CloudConnectorsConfig: CloudConnectorsConfig{
+						LocalRoleARN:  "abc123",
+						GlobalRoleARN: "abc456",
+						ResourceID:    "abc789",
+					},
+				},
+			},
+		},
+	}
+
+	for i, test := range tests {
+		s.Run(fmt.Sprint(i), func() {
+			if test.overwriteEnv != nil {
+				test.overwriteEnv(s.T())
+			}
+			cfg, err := config.NewConfigFrom(test.config)
+			s.Require().NoError(err)
+
+			c, err := New(cfg)
+			s.Require().NoError(err)
+
+			s.Equal(test.expectedType, c.Benchmark)
+			s.Equal(test.expectedCloudConfig, c.CloudConfig)
+		})
+	}
+}
diff --git a/internal/flavors/benchmark/aws.go b/internal/flavors/benchmark/aws.go
index 9b70217cee..1c79ed8aa6 100644
--- a/internal/flavors/benchmark/aws.go
+++ b/internal/flavors/benchmark/aws.go
@@ -84,7 +84,14 @@ func (a *AWS) initialize(ctx context.Context, log *clog.Logger, cfg *config.Conf
 }
 
 func (a *AWS) getIdentity(ctx context.Context, cfg *config.Config) (*awssdk.Config, *cloud.Identity, error) {
-	awsConfig, err := awslib.InitializeAWSConfig(cfg.CloudConfig.Aws.Cred)
+	var awsConfig *awssdk.Config
+	var err error
+
+	if cfg.CloudConfig.Aws.CloudConnectors {
+		awsConfig, err = awslib.InitializeAWSConfigCloudConnectors(ctx, cfg.CloudConfig.Aws)
+	} else {
+		awsConfig, err = awslib.InitializeAWSConfig(cfg.CloudConfig.Aws.Cred)
+	}
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to initialize AWS credentials: %w", err)
 	}
diff --git a/internal/flavors/benchmark/aws_org.go b/internal/flavors/benchmark/aws_org.go
index 4cf8de1483..3f7c51f019 100644
--- a/internal/flavors/benchmark/aws_org.go
+++ b/internal/flavors/benchmark/aws_org.go
@@ -218,7 +218,15 @@ func (a *AWSOrg) pickManagementAccountRole(ctx context.Context, log *clog.Logger
 }
 
 func (a *AWSOrg) getIdentity(ctx context.Context, cfg *config.Config) (*awssdk.Config, *cloud.Identity, error) {
-	awsConfig, err := awslib.InitializeAWSConfig(cfg.CloudConfig.Aws.Cred)
+	var awsConfig *awssdk.Config
+	var err error
+
+	if cfg.CloudConfig.Aws.CloudConnectors {
+		awsConfig, err = awslib.InitializeAWSConfigCloudConnectors(ctx, cfg.CloudConfig.Aws)
+	} else {
+		awsConfig, err = awslib.InitializeAWSConfig(cfg.CloudConfig.Aws.Cred)
+	}
+
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to initialize AWS credentials: %w", err)
 	}
diff --git a/internal/flavors/benchmark/aws_test.go b/internal/flavors/benchmark/aws_test.go
index dd581fca71..b07315c429 100644
--- a/internal/flavors/benchmark/aws_test.go
+++ b/internal/flavors/benchmark/aws_test.go
@@ -21,7 +21,14 @@ import (
 	"errors"
 	"testing"
 
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/credentials/stscreds"
+	libbeataws "github.com/elastic/beats/v7/x-pack/libbeat/common/aws"
+	"github.com/stretchr/testify/mock"
+
 	"github.com/elastic/cloudbeat/internal/config"
+	"github.com/elastic/cloudbeat/internal/dataprovider/providers/cloud"
 	"github.com/elastic/cloudbeat/internal/resources/fetching"
 	"github.com/elastic/cloudbeat/internal/resources/providers/awslib"
 	"github.com/elastic/cloudbeat/internal/resources/utils/testhelper"
@@ -60,6 +67,91 @@ func TestAWS_Initialize(t *testing.T) {
 				fetching.S3Type,
 			},
 		},
+		{
+			name: "cloud connectors",
+			cfg: config.Config{
+				Benchmark: "cis_aws",
+				CloudConfig: config.CloudConfig{
+					Aws: config.AwsConfig{
+						AccountType:     config.SingleAccount,
+						Cred:            libbeataws.ConfigAWS{},
+						CloudConnectors: true,
+						CloudConnectorsConfig: config.CloudConnectorsConfig{
+							LocalRoleARN:  "abc123",
+							GlobalRoleARN: "abc456",
+							ResourceID:    "abc789",
+						},
+					},
+				},
+			},
+			identityProvider: func() awslib.IdentityProviderGetter {
+				cfgMatcher := mock.MatchedBy(func(cfg aws.Config) bool {
+					c, is := cfg.Credentials.(*aws.CredentialsCache)
+					if !is {
+						return false
+					}
+					return c.IsCredentialsProvider(&stscreds.AssumeRoleProvider{})
+				})
+				identityProvider := &awslib.MockIdentityProviderGetter{}
+				identityProvider.EXPECT().GetIdentity(mock.Anything, cfgMatcher).Return(
+					&cloud.Identity{
+						Account: "test-account",
+					},
+					nil,
+				)
+
+				return identityProvider
+			}(),
+			want: []string{
+				fetching.IAMType,
+				fetching.KmsType,
+				fetching.TrailType,
+				fetching.AwsMonitoringType,
+				fetching.EC2NetworkingType,
+				fetching.RdsType,
+				fetching.S3Type,
+			},
+		},
+		{
+			name: "no credential cache in non cloud connectors setup",
+			cfg: config.Config{
+				Benchmark: "cis_aws",
+				CloudConfig: config.CloudConfig{
+					Aws: config.AwsConfig{
+						AccountType: config.SingleAccount,
+						Cred: libbeataws.ConfigAWS{
+							AccessKeyID:     "keyid",
+							SecretAccessKey: "key",
+						},
+						CloudConnectors: false,
+					},
+				},
+			},
+			identityProvider: func() awslib.IdentityProviderGetter {
+				cfgMatcher := mock.MatchedBy(func(cfg aws.Config) bool {
+					_, is := cfg.Credentials.(credentials.StaticCredentialsProvider)
+					return is
+				})
+				identityProvider := &awslib.MockIdentityProviderGetter{}
+				identityProvider.EXPECT().GetIdentity(mock.Anything, cfgMatcher).Return(
+					&cloud.Identity{
+						Account: "test-account",
+					},
+					nil,
+				)
+
+				return identityProvider
+			}(),
+			want: []string{
+				fetching.IAMType,
+				fetching.KmsType,
+				fetching.TrailType,
+				fetching.AwsMonitoringType,
+				fetching.EC2NetworkingType,
+				fetching.RdsType,
+				fetching.S3Type,
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
diff --git a/internal/resources/providers/awslib/config.go b/internal/resources/providers/awslib/config.go
index 0c7c08d333..d30b58f648 100644
--- a/internal/resources/providers/awslib/config.go
+++ b/internal/resources/providers/awslib/config.go
@@ -18,28 +18,106 @@
 package awslib
 
 import (
+	"context"
+	"fmt"
 	"net/http"
+	"time"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/aws/retry"
+	awsconfig "github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials/stscreds"
+	"github.com/aws/aws-sdk-go-v2/service/sts"
 	libbeataws "github.com/elastic/beats/v7/x-pack/libbeat/common/aws"
+
+	"github.com/elastic/cloudbeat/internal/config"
 )
 
+func awsConfigRetrier() aws.Retryer {
+	return retry.NewStandard(func(o *retry.StandardOptions) {
+		o.Retryables = append(o.Retryables, retry.RetryableHTTPStatusCode{
+			Codes: map[int]struct{}{
+				http.StatusTooManyRequests: {},
+			},
+		})
+	})
+}
+
 func InitializeAWSConfig(cfg libbeataws.ConfigAWS) (*aws.Config, error) {
 	awsConfig, err := libbeataws.InitializeAWSConfig(cfg)
 	if err != nil {
 		return nil, err
 	}
 
-	awsConfig.Retryer = func() aws.Retryer {
-		return retry.NewStandard(func(o *retry.StandardOptions) {
-			o.Retryables = append(o.Retryables, retry.RetryableHTTPStatusCode{
-				Codes: map[int]struct{}{
-					http.StatusTooManyRequests: {},
-				},
-			})
-		})
-	}
+	awsConfig.Retryer = awsConfigRetrier
 
 	return &awsConfig, nil
 }
+
+func CloudConnectorsExternalID(resourceID, externalIDPart string) string {
+	return fmt.Sprintf("%s-%s", resourceID, externalIDPart)
+}
+
+func InitializeAWSConfigCloudConnectors(ctx context.Context, cfg config.AwsConfig) (*aws.Config, error) {
+	const defaultDuration = 20 * time.Minute
+
+	// 1. Load initial config - Chain Step 1 - Elastic Super Role Local implicitly assumed through IRSA.
+	awsConfig, err := awsconfig.LoadDefaultConfig(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	chain := []AWSRoleChainingStep{
+		// Chain Step 2 - Elastic Super Role Global
+		{
+			RoleARN: cfg.CloudConnectorsConfig.GlobalRoleARN,
+			Options: func(aro *stscreds.AssumeRoleOptions) {
+				aro.RoleSessionName = "cloudbeat-super-role-global"
+				aro.Duration = defaultDuration
+			},
+		},
+		// Chain Step 3 - Elastic Super Role Local
+		{
+			RoleARN: cfg.Cred.RoleArn,
+			Options: func(aro *stscreds.AssumeRoleOptions) {
+				aro.RoleSessionName = "cloudbeat-remote-role"
+				aro.Duration = cfg.Cred.AssumeRoleDuration
+				aro.ExternalID = aws.String(CloudConnectorsExternalID(cfg.CloudConnectorsConfig.ResourceID, cfg.Cred.ExternalID))
+			},
+		},
+	}
+
+	retConf := AWSConfigRoleChaining(awsConfig, chain)
+	retConf.Retryer = awsConfigRetrier
+
+	return retConf, nil
+}
+
+// AWSConfigRoleChaining initializes an assume role provider and an credential cache for each step on the chain, using the previous one as client.
+func AWSConfigRoleChaining(initialConfig aws.Config, chain []AWSRoleChainingStep) *aws.Config {
+	var client *sts.Client
+	var assumeRoleProvider *stscreds.AssumeRoleProvider
+	var credentialsCache *aws.CredentialsCache
+	cnf := initialConfig
+
+	for _, c := range chain {
+		client = sts.NewFromConfig(cnf) // create client using the credentials from previous or initial step.
+
+		// create a assume role provider for the current chain part role.
+		assumeRoleProvider = stscreds.NewAssumeRoleProvider(
+			client,
+			c.RoleARN,
+			c.Options,
+		)
+		credentialsCache = aws.NewCredentialsCache(assumeRoleProvider)
+
+		cnf.Credentials = credentialsCache
+	}
+
+	return &cnf
+}
+
+type AWSRoleChainingStep struct {
+	RoleARN string
+	Options func(aro *stscreds.AssumeRoleOptions)
+}
diff --git a/scripts/packaging/docker/elastic-agent/Dockerfile b/scripts/packaging/docker/elastic-agent/Dockerfile
index 700312bc38..72fe7957e8 100644
--- a/scripts/packaging/docker/elastic-agent/Dockerfile
+++ b/scripts/packaging/docker/elastic-agent/Dockerfile
@@ -1,6 +1,6 @@
 ARG ELASTIC_AGENT_IMAGE=docker.elastic.co/beats/elastic-agent:8.14.0-SNAPSHOT
 
-FROM ${ELASTIC_AGENT_IMAGE} as elastic_agent_cloudbeat
+FROM ${ELASTIC_AGENT_IMAGE} AS elastic_agent_cloudbeat
 COPY --chown=elastic-agent:elastic-agent --chmod=755 cloudbeat /tmp/components/cloudbeat
 COPY --chown=elastic-agent:elastic-agent --chmod=666 bundle.tar.gz /tmp/components/bundle.tar.gz
 COPY --chown=elastic-agent:elastic-agent --chmod=644 cloudbeat.yml /tmp/components/cloudbeat.yml