feat(cloudtrail): support accounts for org trails

plugins/cloudtrail/README.md

@@ -130,6 +130,7 @@ The json object has the following properties:
 * `s3DownloadConcurrency`: value is numeric. Controls the number of background goroutines used to download S3 files. (Default: 1)
 * `S3Interval`: value is string. Download log files matching the specified time interval. Note that this matches log file *names*, not event timestamps. CloudTrail logs usually cover [the previous 5 minutes of activity](https://docs.aws.amazon.com/awscloudtrail/latest/userguide/get-and-view-cloudtrail-log-files.html). See *Time Intervals* below for possible formats.
 * `useS3SNS`: value is boolean. If true, then the plugin will expect SNS messages to originate from S3 instead of directly from Cloudtrail (Default: false)
+* `S3AccountList`: value is string. Download log files matching the specified account IDs (in a comma separated list) in an organization trail. See *Read From S3 Bucket Directly* below for more details.
 
 The init string can be the empty string, which is treated identically to `{}`.
 
@@ -165,6 +166,10 @@ All objects below the bucket, or below the bucket + prefix, will be considered c
 
 For example, if a bucket `my-s3-bucket` contained cloudtrail logs below a prefix `AWSLogs/411571310278/CloudTrail/us-west-1/2021/09/23/`, Using an open params of `s3://my-s3-bucket/AWSLogs/411571310278/CloudTrail/us-west-1/2021/09/23/` would configure the plugin to read all files below `AWSLogs/411571310278/CloudTrail/us-west-1/2021/09/23/` as cloudtrail logs and then return EOF. No other files in the bucket will be read.
 
+For organization trails the files are normally stored like `s3://bucket_name/prefix_name/AWSLogs/O-ID/Account ID/CloudTrail/Region/YYYY/MM/DD/file_name.json.gz`. Using an open parameter of `s3//my-s3-bucket/AWSLogs/o-123abc/` would configure the plugin to read all files for all account IDs in the organization `o-123abc`, for all regions and the entire retention time. Therefore it makes sense to combine this open parameter with `S3AccountList` and `S3Interval` parameters. `S3AccountList` is a comma separated string with account IDs to query.
+
+Setting `S3AccountList` to `012345678912,987654321012` and `S3Interval` to `3d-1d` with open parameter `s3://my-s3-bucket/AWSLogs/o-123abc/` would get all events for account IDs 12345678912 and 987654321012 for all regions from 3 days ago up to to 1 day ago.
+
 #### Read from SQS Queue
 
 When using `sqs://<SQS Queue Name>`, the plugin will read messages from the provided SQS Queue. The messages are assumed to be [SNS Notifications](https://docs.aws.amazon.com/awscloudtrail/latest/userguide/configure-sns-notifications-for-cloudtrail.html) that announce the presence of new Cloudtrail log files in a S3 bucket. Each new file will be read from the provided s3 bucket.

plugins/cloudtrail/pkg/cloudtrail/cloudtrail.go

@@ -48,7 +48,7 @@ const (
 	PluginName               = "cloudtrail"
 	PluginDescription        = "reads cloudtrail JSON data saved to file in the directory specified in the settings"
 	PluginContact            = "github.com/falcosecurity/plugins/"
-	PluginVersion            = "0.11.0"
+	PluginVersion            = "0.12.0"
 	PluginEventSource        = "aws_cloudtrail"
 )
 

plugins/cloudtrail/pkg/cloudtrail/config.go

@@ -24,6 +24,7 @@ type PluginConfig struct {
 	SQSDelete             bool            `json:"sqsDelete" jsonschema:"title=Delete SQS messages,description=If true then the plugin will delete SQS messages from the queue immediately after receiving them (Default: true),default=true"`
 	UseAsync              bool            `json:"useAsync" jsonschema:"title=Use async extraction,description=If true then async extraction optimization is enabled (Default: true),default=true"`
 	UseS3SNS              bool            `json:"useS3SNS" jsonschema:"title=Use S3 SNS,description=If true then the plugin will expect SNS messages to originate from S3 instead of directly from Cloudtrail (Default: false),default=false"`
+	S3AccountList         string          `json:"s3AccountList" jsonschema:"title=S3 account list,description=A comma separated list of account IDs for organizational Cloudtrails (Default: no account IDs),default="`
 	AWS                   PluginConfigAWS `json:"aws"`
 }
 
@@ -34,5 +35,6 @@ func (p *PluginConfig) Reset() {
 	p.S3Interval = ""
 	p.UseAsync = true
 	p.UseS3SNS = false
+	p.S3AccountList = ""
 	p.AWS.Reset()
 }

plugins/cloudtrail/pkg/cloudtrail/source.go

@@ -182,13 +182,20 @@ func (oCtx *PluginInstance) openS3(input string) error {
 
 	var inputParams []listOrigin
 	ctx := context.Background()
+	var intervalPrefixList []string
 
 	startTime, endTime, err := ParseInterval(oCtx.config.S3Interval)
 	if err != nil {
 		return fmt.Errorf(PluginName + " invalid interval: \"%s\": %s", oCtx.config.S3Interval, err.Error())
 
 	}
 
+	s3AccountList := oCtx.config.S3AccountList
+	accountListRE := regexp.MustCompile(`^(?: *\d{12} *,?)*$`)
+	if (! accountListRE.MatchString(s3AccountList)) {
+		return fmt.Errorf(PluginName + " invalid account list: \"%s\"", oCtx.config.S3AccountList)
+}
+
 	// CloudTrail logs have the format
 	// bucket_name/prefix_name/AWSLogs/Account ID/CloudTrail/region/YYYY/MM/DD/AccountID_CloudTrail_RegionName_YYYYMMDDTHHmmZ_UniqueString.json.gz
 	// for organization trails the format is
@@ -200,32 +207,86 @@ func (oCtx *PluginInstance) openS3(input string) error {
 
 	// For durations, carve out a special case for "Copy S3 URI" in the AWS console, which gives you
 	// bucket_name/prefix_name/AWSLogs/<Account ID>/ or bucket_name/prefix_name/AWSLogs/<Org-ID>/<Account ID>/
-	awsLogsRE := regexp.MustCompile(`AWSLogs/(?:o-[a-z0-9]{10,32}/)?\d+/?$`)
+	awsLogsRE := regexp.MustCompile(`AWSLogs/(?:o-[a-z0-9]{10,32}/)?\d{12}/?$`)
+	awsLogsOrgRE := regexp.MustCompile(`AWSLogs/o-[a-z0-9]{10,32}/?$`)
 	if awsLogsRE.MatchString(prefix) {
 		if (! strings.HasSuffix(intervalPrefix, "/")) {
 			intervalPrefix += "/"
 		}
 		intervalPrefix += "CloudTrail/"
-	}
+		intervalPrefixList = append(intervalPrefixList, intervalPrefix)
+	} else if awsLogsOrgRE.MatchString(prefix) {
+		if (! strings.HasSuffix(intervalPrefix, "/")) {
+			intervalPrefix += "/"
+		}
+		if s3AccountList != "" {
+			// build intervalPrefixList by using the provided S3AccountList
+			accountListArray := strings.Split(s3AccountList , ",")
+			if len(accountListArray) <= 0 {
+				return fmt.Errorf(PluginName + " invalid account list: \"%s\"", oCtx.config.S3AccountList)
+			}
+			for i := range accountListArray {
+				accountListArray[i] = strings.TrimSpace(accountListArray[i])
+			}
+			for _, account := range accountListArray {
+				intervalPrefixList = append(intervalPrefixList, intervalPrefix + account + "/CloudTrail/")
+			}
+		} else {
+			// try to get all available account IDs in the S3 CloudTrail bucket
+			delimiter := "/"
+			paginator := s3.NewListObjectsV2Paginator(oCtx.s3.client, &s3.ListObjectsV2Input{
+				Bucket: &oCtx.s3.bucket,
+				Prefix: &intervalPrefix,
+				Delimiter: &delimiter,
+			})
+			for paginator.HasMorePages() {
+				page, err := paginator.NextPage(ctx)
+				if err != nil {
+					// Try friendlier error sources first.
+					var aErr smithy.APIError
+					if errors.As(err, &aErr) {
+						return fmt.Errorf(PluginName + " plugin error: %s: %s", aErr.ErrorCode(), aErr.ErrorMessage())
+					}
 
-	if strings.HasSuffix(intervalPrefix, "/CloudTrail/") {
-		delimiter := "/"
-		// Fetch the list of regions.
-		output, err := oCtx.s3.client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
-			Bucket: &oCtx.s3.bucket,
-			Prefix: &intervalPrefix,
-			Delimiter: &delimiter,
-		})
-		if err == nil {
-			for _, commonPrefix := range output.CommonPrefixes {
-				params := listOrigin {prefix: commonPrefix.Prefix}
-				if !startTime.IsZero() {
-					// startAfter doesn't have to be a real key.
-					startAfterSuffix := startTime.Format("2006/01/02/")
-					startAfter := *commonPrefix.Prefix + startAfterSuffix
-					params.startAfter = &startAfter
+					var oErr *smithy.OperationError
+					if errors.As(err, &oErr) {
+						return fmt.Errorf(PluginName + " plugin error: %s: %s", oErr.Service(), oErr.Unwrap())
+					}
+
+					return fmt.Errorf(PluginName + " plugin error: failed to list accounts: " + err.Error())
+				}
+				for _, commonPrefix := range page.CommonPrefixes {
+					path := commonPrefix.Prefix
+					if awsLogsRE.MatchString(*path) {
+						intervalPrefixList = append(intervalPrefixList, *path + "CloudTrail/")
+					}
+				}
+			}
+		}
+	} else {
+		intervalPrefixList = append(intervalPrefixList, intervalPrefix)
+	}
+
+	for _, intervalPrefix := range intervalPrefixList {
+		if strings.HasSuffix(intervalPrefix, "/CloudTrail/") {
+			delimiter := "/"
+			// Fetch the list of regions.
+			output, err := oCtx.s3.client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
+				Bucket: &oCtx.s3.bucket,
+				Prefix: &intervalPrefix,
+				Delimiter: &delimiter,
+			})
+			if err == nil {
+				for _, commonPrefix := range output.CommonPrefixes {
+					params := listOrigin {prefix: commonPrefix.Prefix}
+					if !startTime.IsZero() {
+						// startAfter doesn't have to be a real key.
+						startAfterSuffix := startTime.Format("2006/01/02/")
+						startAfter := *commonPrefix.Prefix + startAfterSuffix
+						params.startAfter = &startAfter
+					}
+					inputParams = append(inputParams, params)
 				}
-				inputParams = append(inputParams, params)
 			}
 		}
 	}