Status Handler; Expose aws permission issues to user using beats status. (#3515)

* Status Handler; Expose aws permission issues to user using beats status.

* more tests

* statushandler tests

* fix test lint

* fix test

* linter fixes

* fix docker file

* cleanup

* testcases

* aws org flow: filter only the valid accounts

* generate mocks

* expand to asset inventory

* change MultiRegionFetch error wrapping

* fixes

* nits

Author: moukoublen

internal/flavors/asset_inventory.go

@@ -28,6 +28,7 @@ import (
 	"github.com/elastic/cloudbeat/internal/flavors/assetinventory"
 	"github.com/elastic/cloudbeat/internal/infra/clog"
 	"github.com/elastic/cloudbeat/internal/inventory"
+	"github.com/elastic/cloudbeat/internal/statushandler"
 )
 
 type assetInventory struct {
@@ -53,7 +54,9 @@ func newAssetInventoryFromCfg(b *beat.Beat, cfg *config.Config) (*assetInventory
 		return nil, fmt.Errorf("failed to init client: %w", err)
 	}
 
-	strategy := assetinventory.GetStrategy(logger, cfg)
+	statusHandler := statushandler.NewStatusHandler(b.Manager)
+
+	strategy := assetinventory.GetStrategy(logger, cfg, statusHandler)
 	newAssetInventory, err := strategy.NewAssetInventory(ctx, beatClient)
 	if err != nil {
 		cancel()

internal/flavors/assetinventory/strategy.go

@@ -36,15 +36,17 @@ import (
 	gcp_auth "github.com/elastic/cloudbeat/internal/resources/providers/gcplib/auth"
 	gcp_inventory "github.com/elastic/cloudbeat/internal/resources/providers/gcplib/inventory"
 	"github.com/elastic/cloudbeat/internal/resources/providers/msgraph"
+	"github.com/elastic/cloudbeat/internal/statushandler"
 )
 
 type Strategy interface {
 	NewAssetInventory(ctx context.Context, client beat.Client) (inventory.AssetInventory, error)
 }
 
 type strategy struct {
-	logger *clog.Logger
-	cfg    *config.Config
+	logger        *clog.Logger
+	cfg           *config.Config
+	statusHandler statushandler.StatusHandlerAPI
 }
 
 func (s *strategy) NewAssetInventory(ctx context.Context, client beat.Client) (inventory.AssetInventory, error) {
@@ -55,7 +57,7 @@ func (s *strategy) NewAssetInventory(ctx context.Context, client beat.Client) (i
 	case config.ProviderAWS:
 		switch s.cfg.CloudConfig.Aws.AccountType {
 		case config.SingleAccount, config.OrganizationAccount:
-			fetchers, err = s.initAwsFetchers(ctx)
+			fetchers, err = s.initAwsFetchers(ctx, s.statusHandler)
 		default:
 			err = fmt.Errorf("unsupported account_type: %q", s.cfg.CloudConfig.Aws.AccountType)
 		}
@@ -111,9 +113,10 @@ func (s *strategy) initGcpFetchers(ctx context.Context) ([]inventory.AssetFetche
 	return gcpfetcher.New(s.logger, provider), nil
 }
 
-func GetStrategy(logger *clog.Logger, cfg *config.Config) Strategy {
+func GetStrategy(logger *clog.Logger, cfg *config.Config, statusHandler statushandler.StatusHandlerAPI) Strategy {
 	return &strategy{
-		logger: logger,
-		cfg:    cfg,
+		logger:        logger,
+		cfg:           cfg,
+		statusHandler: statusHandler,
 	}
 }

internal/flavors/assetinventory/strategy_aws.go

@@ -20,19 +20,16 @@ package assetinventory
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	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/s3"
 	"github.com/aws/aws-sdk-go-v2/service/sts"
 
 	"github.com/elastic/cloudbeat/internal/config"
-	"github.com/elastic/cloudbeat/internal/infra/clog"
 	"github.com/elastic/cloudbeat/internal/inventory"
 	"github.com/elastic/cloudbeat/internal/inventory/awsfetcher"
 	"github.com/elastic/cloudbeat/internal/resources/providers/awslib"
-	"github.com/elastic/cloudbeat/internal/resources/utils/pointers"
+	"github.com/elastic/cloudbeat/internal/statushandler"
 )
 
 const (
@@ -48,7 +45,7 @@ func (s *strategy) getInitialAWSConfig(ctx context.Context, cfg *config.Config)
 	return awslib.InitializeAWSConfig(cfg.CloudConfig.Aws.Cred, s.logger)
 }
 
-func (s *strategy) initAwsFetchers(ctx context.Context) ([]inventory.AssetFetcher, error) {
+func (s *strategy) initAwsFetchers(ctx context.Context, statusHandler statushandler.StatusHandlerAPI) ([]inventory.AssetFetcher, error) {
 	awsConfig, err := s.getInitialAWSConfig(ctx, s.cfg)
 	if err != nil {
 		return nil, err
@@ -62,7 +59,7 @@ func (s *strategy) initAwsFetchers(ctx context.Context) ([]inventory.AssetFetche
 
 	// Early exit if we're scanning the entire account.
 	if s.cfg.CloudConfig.Aws.AccountType == config.SingleAccount {
-		return awsfetcher.New(ctx, s.logger, awsIdentity, *awsConfig), nil
+		return awsfetcher.New(ctx, s.logger, awsIdentity, *awsConfig, statusHandler), nil
 	}
 
 	// Assume audit roles per selected account and generate fetchers for them
@@ -84,30 +81,18 @@ func (s *strategy) initAwsFetchers(ctx context.Context) ([]inventory.AssetFetche
 			rootRoleConfig,
 			fmtIAMRole(identity.Account, memberRole),
 		)
-		if ok := tryListingBuckets(ctx, s.logger, assumedRoleConfig); !ok {
+		if ok := awslib.CredentialsValid(ctx, assumedRoleConfig, s.logger); !ok {
 			// role does not exist, skip identity/account
 			s.logger.Infof("Skipping identity on purpose %+v", identity)
 			continue
 		}
-		accountFetchers := awsfetcher.New(ctx, s.logger, &identity, assumedRoleConfig)
+		accountFetchers := awsfetcher.New(ctx, s.logger, &identity, assumedRoleConfig, statusHandler)
 		fetchers = append(fetchers, accountFetchers...)
 	}
 
 	return fetchers, nil
 }
 
-func tryListingBuckets(ctx context.Context, log *clog.Logger, roleConfig awssdk.Config) bool {
-	s3Client := s3.NewFromConfig(roleConfig)
-	_, err := s3Client.ListBuckets(ctx, &s3.ListBucketsInput{MaxBuckets: pointers.Ref(int32(1))})
-	if err == nil {
-		return true
-	}
-	if !strings.Contains(err.Error(), "not authorized to perform: sts:AssumeRole") {
-		log.Errorf("Expected a 403 autorization error, but got: %v", err)
-	}
-	return false
-}
-
 func assumeRole(client stscreds.AssumeRoleAPIClient, cfg awssdk.Config, arn string) awssdk.Config {
 	cfg.Credentials = awssdk.NewCredentialsCache(stscreds.NewAssumeRoleProvider(client, arn))
 	return cfg

internal/flavors/benchmark/aws.go

@@ -33,12 +33,14 @@ import (
 	"github.com/elastic/cloudbeat/internal/resources/fetching/preset"
 	"github.com/elastic/cloudbeat/internal/resources/fetching/registry"
 	"github.com/elastic/cloudbeat/internal/resources/providers/awslib"
+	"github.com/elastic/cloudbeat/internal/statushandler"
 )
 
 const resourceChBufferSize = 10000
 
 type AWS struct {
 	IdentityProvider awslib.IdentityProviderGetter
+	StatusHandler    statushandler.StatusHandlerAPI
 }
 
 func (a *AWS) NewBenchmark(ctx context.Context, log *clog.Logger, cfg *config.Config) (builder.Benchmark, error) {
@@ -50,7 +52,7 @@ func (a *AWS) NewBenchmark(ctx context.Context, log *clog.Logger, cfg *config.Co
 
 	return builder.New(
 		builder.WithBenchmarkDataProvider(bdp),
-	).Build(ctx, log, cfg, resourceCh, reg)
+	).Build(ctx, log, cfg, resourceCh, reg, a.StatusHandler)
 }
 
 //revive:disable-next-line:function-result-limit
@@ -79,7 +81,7 @@ func (a *AWS) initialize(ctx context.Context, log *clog.Logger, cfg *config.Conf
 
 	return registry.NewRegistry(
 		log,
-		registry.WithFetchersMap(preset.NewCisAwsFetchers(ctx, log, *awsConfig, ch, awsIdentity)),
+		registry.WithFetchersMap(preset.NewCisAwsFetchers(ctx, log, *awsConfig, ch, awsIdentity, a.StatusHandler)),
 	), cloud.NewDataProvider(cloud.WithAccount(*awsIdentity)), nil, nil
 }
 

internal/flavors/benchmark/aws_org.go

@@ -26,6 +26,7 @@ import (
 	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/samber/lo"
 	"go.opentelemetry.io/otel"
 
 	"github.com/elastic/cloudbeat/internal/config"
@@ -40,6 +41,7 @@ import (
 	"github.com/elastic/cloudbeat/internal/resources/providers/awslib"
 	"github.com/elastic/cloudbeat/internal/resources/providers/awslib/iam"
 	"github.com/elastic/cloudbeat/internal/resources/utils/pointers"
+	"github.com/elastic/cloudbeat/internal/statushandler"
 )
 
 const (
@@ -53,9 +55,11 @@ const (
 var tracer = otel.Tracer(scopeName)
 
 type AWSOrg struct {
-	IAMProvider      iam.RoleGetter
-	IdentityProvider awslib.IdentityProviderGetter
-	AccountProvider  awslib.AccountProviderAPI
+	IAMProvider       iam.RoleGetter
+	IdentityProvider  awslib.IdentityProviderGetter
+	AccountProvider   awslib.AccountProviderAPI
+	StatusHandler     statushandler.StatusHandlerAPI
+	AWSCredsValidator awslib.CredentialsValidator
 }
 
 func (a *AWSOrg) NewBenchmark(ctx context.Context, log *clog.Logger, cfg *config.Config) (builder.Benchmark, error) {
@@ -67,7 +71,7 @@ func (a *AWSOrg) NewBenchmark(ctx context.Context, log *clog.Logger, cfg *config
 
 	return builder.New(
 		builder.WithBenchmarkDataProvider(bdp),
-	).Build(ctx, log, cfg, resourceCh, reg)
+	).Build(ctx, log, cfg, resourceCh, reg, a.StatusHandler)
 }
 
 //revive:disable-next-line:function-result-limit
@@ -110,7 +114,13 @@ func (a *AWSOrg) initialize(ctx context.Context, log *clog.Logger, cfg *config.C
 				return nil, observability.FailSpan(span, "failed to get AWS accounts", err)
 			}
 
-			fm := preset.NewCisAwsOrganizationFetchers(ctx, spannedLog, ch, accounts, cache)
+			// Filter the accounts to the ones having valid credentials on each aws account.
+			// Meaning only the accounts that have the security audit role created and thus were selected by customer on cloud formation.
+			filtered := lo.Filter(accounts, func(item preset.AwsAccount, _ int) bool {
+				return a.AWSCredsValidator.Validate(ctx, item.Config, log)
+			})
+
+			fm := preset.NewCisAwsOrganizationFetchers(ctx, spannedLog, ch, filtered, cache, a.StatusHandler)
 			m := make(registry.FetchersMap)
 			for accountId, fetchersMap := range fm {
 				for key, fetcher := range fetchersMap {

internal/flavors/benchmark/aws_org_test.go

@@ -40,6 +40,7 @@ import (
 	"github.com/elastic/cloudbeat/internal/resources/providers/awslib/iam"
 	"github.com/elastic/cloudbeat/internal/resources/utils/pointers"
 	"github.com/elastic/cloudbeat/internal/resources/utils/testhelper"
+	"github.com/elastic/cloudbeat/internal/statushandler"
 )
 
 var expectedAWSSubtypes = []string{
@@ -116,9 +117,11 @@ func TestAWSOrg_Initialize(t *testing.T) {
 			t.Parallel()
 
 			testInitialize(t, &AWSOrg{
-				IAMProvider:      tt.iamProvider,
-				IdentityProvider: tt.identityProvider,
-				AccountProvider:  tt.accountProvider,
+				IAMProvider:       tt.iamProvider,
+				IdentityProvider:  tt.identityProvider,
+				AccountProvider:   tt.accountProvider,
+				StatusHandler:     statushandler.NewMockStatusHandlerAPI(t),
+				AWSCredsValidator: awslib.CredentialsValidatorNOOP,
 			}, &tt.cfg, tt.wantErr, tt.want)
 		})
 	}
@@ -168,9 +171,11 @@ func Test_getAwsAccounts(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			a := AWSOrg{
-				IAMProvider:      getMockIAMRoleGetter([]iam.Role{*makeRole("cloudbeat-root")}),
-				IdentityProvider: nil,
-				AccountProvider:  tt.accountProvider,
+				IAMProvider:       getMockIAMRoleGetter([]iam.Role{*makeRole("cloudbeat-root")}),
+				IdentityProvider:  nil,
+				AccountProvider:   tt.accountProvider,
+				StatusHandler:     statushandler.NewMockStatusHandlerAPI(t),
+				AWSCredsValidator: awslib.CredentialsValidatorNOOP,
 			}
 			log := testhelper.NewLogger(t)
 			got, err := a.getAwsAccounts(t.Context(), log, aws.Config{}, &tt.rootIdentity)
@@ -245,9 +250,11 @@ func Test_pickManagementAccountRole(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			a := AWSOrg{
-				IAMProvider:      getMockIAMRoleGetter(tt.roles),
-				IdentityProvider: mockAwsIdentityProvider(nil),
-				AccountProvider:  mockAccountProvider(nil),
+				IAMProvider:       getMockIAMRoleGetter(tt.roles),
+				IdentityProvider:  mockAwsIdentityProvider(nil),
+				AccountProvider:   mockAccountProvider(nil),
+				StatusHandler:     statushandler.NewMockStatusHandlerAPI(t),
+				AWSCredsValidator: awslib.CredentialsValidatorNOOP,
 			}
 
 			// set up log capture

internal/flavors/benchmark/aws_test.go

@@ -32,6 +32,7 @@ import (
 	"github.com/elastic/cloudbeat/internal/resources/fetching"
 	"github.com/elastic/cloudbeat/internal/resources/providers/awslib"
 	"github.com/elastic/cloudbeat/internal/resources/utils/testhelper"
+	"github.com/elastic/cloudbeat/internal/statushandler"
 )
 
 func TestAWS_Initialize(t *testing.T) {
@@ -159,6 +160,7 @@ func TestAWS_Initialize(t *testing.T) {
 
 			testInitialize(t, &AWS{
 				IdentityProvider: tt.identityProvider,
+				StatusHandler:    statushandler.NewMockStatusHandlerAPI(t),
 			}, &tt.cfg, tt.wantErr, tt.want)
 		})
 	}

internal/flavors/benchmark/azure.go

@@ -34,6 +34,7 @@ import (
 	"github.com/elastic/cloudbeat/internal/resources/fetching/registry"
 	"github.com/elastic/cloudbeat/internal/resources/providers/azurelib"
 	"github.com/elastic/cloudbeat/internal/resources/providers/azurelib/auth"
+	"github.com/elastic/cloudbeat/internal/statushandler"
 )
 
 type Azure struct {
@@ -51,7 +52,7 @@ func (a *Azure) NewBenchmark(ctx context.Context, log *clog.Logger, cfg *config.
 	return builder.New(
 		builder.WithBenchmarkDataProvider(bdp),
 		builder.WithManagerTimeout(calculateFetcherTimeout(cfg.Period)),
-	).Build(ctx, log, cfg, resourceCh, reg)
+	).Build(ctx, log, cfg, resourceCh, reg, statushandler.NOOP{})
 }
 
 //revive:disable-next-line:function-result-limit

internal/flavors/benchmark/builder/builder.go

@@ -32,6 +32,7 @@ import (
 	"github.com/elastic/cloudbeat/internal/resources/fetching"
 	"github.com/elastic/cloudbeat/internal/resources/fetching/manager"
 	"github.com/elastic/cloudbeat/internal/resources/fetching/registry"
+	"github.com/elastic/cloudbeat/internal/statushandler"
 	"github.com/elastic/cloudbeat/internal/transformer"
 	"github.com/elastic/cloudbeat/version"
 )
@@ -63,12 +64,12 @@ func New(options ...Option) *Builder {
 	return b
 }
 
-func (b *Builder) Build(ctx context.Context, log *clog.Logger, cfg *config.Config, resourceCh chan fetching.ResourceInfo, reg registry.Registry) (Benchmark, error) {
-	return b.buildBase(ctx, log, cfg, resourceCh, reg)
+func (b *Builder) Build(ctx context.Context, log *clog.Logger, cfg *config.Config, resourceCh chan fetching.ResourceInfo, reg registry.Registry, statusHandler statushandler.StatusHandlerAPI) (Benchmark, error) {
+	return b.buildBase(ctx, log, cfg, resourceCh, reg, statusHandler)
 }
 
-func (b *Builder) buildBase(ctx context.Context, log *clog.Logger, cfg *config.Config, resourceCh chan fetching.ResourceInfo, reg registry.Registry) (*basebenchmark, error) {
-	manager, err := manager.NewManager(ctx, log, cfg.Period, b.managerTimeout, reg)
+func (b *Builder) buildBase(ctx context.Context, log *clog.Logger, cfg *config.Config, resourceCh chan fetching.ResourceInfo, reg registry.Registry, statusHandler statushandler.StatusHandlerAPI) (*basebenchmark, error) {
+	manager, err := manager.NewManager(ctx, log, cfg.Period, b.managerTimeout, reg, statusHandler)
 	if err != nil {
 		return nil, err
 	}

internal/flavors/benchmark/builder/builder_test.go

@@ -31,6 +31,7 @@ import (
 	"github.com/elastic/cloudbeat/internal/resources/fetching"
 	"github.com/elastic/cloudbeat/internal/resources/fetching/registry"
 	"github.com/elastic/cloudbeat/internal/resources/utils/testhelper"
+	"github.com/elastic/cloudbeat/internal/statushandler"
 	"github.com/elastic/cloudbeat/internal/uniqueness"
 )
 
@@ -63,12 +64,15 @@ func TestBase_Build_Success(t *testing.T) {
 			path, err := filepath.Abs("../../../../bundle.tar.gz")
 			require.NoError(t, err)
 
+			sh := statushandler.NewMockStatusHandlerAPI(t)
+			sh.EXPECT().Reset().Once()
+
 			resourceCh := make(chan fetching.ResourceInfo)
 			reg := registry.NewMockRegistry(t)
 			benchmark, err := New(tt.opts...).Build(t.Context(), log, &config.Config{
 				BundlePath: path,
 				Period:     time.Minute,
-			}, resourceCh, reg)
+			}, resourceCh, reg, sh)
 			require.NoError(t, err)
 			assert.IsType(t, tt.benchType, benchmark)