cloudquery · kodiakhq · Feb 22, 2023 · Feb 22, 2023 · Feb 22, 2023 · Feb 22, 2023
@@ -19,6 +19,7 @@ import (
 	"github.com/aws/smithy-go"
 	"github.com/aws/smithy-go/logging"
 	"github.com/cloudquery/cloudquery/plugins/source/aws/client/services"
+	"github.com/cloudquery/plugin-sdk/backend"
 	"github.com/cloudquery/plugin-sdk/plugins/source"
 	"github.com/cloudquery/plugin-sdk/schema"
 	"github.com/cloudquery/plugin-sdk/specs"
@@ -37,6 +38,7 @@ type Client struct {
 	WAFScope             wafv2types.Scope
 	Partition            string
 	LanguageCode         string
+	Backend              backend.Backend
 }
 
 type AwsLogger struct {
@@ -103,8 +105,9 @@ func (s *ServicesManager) InitServicesForPartitionAccountAndScope(partition, acc
 	s.wafScopeServices[partition][accountId] = &svcs
 }
 
-func NewAwsClient(logger zerolog.Logger) Client {
+func NewAwsClient(logger zerolog.Logger, b backend.Backend) Client {
 	return Client{
+		Backend: b,
 		ServicesManager: ServicesManager{
 			services: ServicesPartitionAccountRegionMap{},
 		},
@@ -145,6 +148,7 @@ func (c *Client) withPartitionAccountIDAndRegion(partition, accountID, region st
 		Region:               region,
 		AutoscalingNamespace: c.AutoscalingNamespace,
 		WAFScope:             c.WAFScope,
+		Backend:              c.Backend,
 	}
 }
 
@@ -157,6 +161,7 @@ func (c *Client) withPartitionAccountIDRegionAndNamespace(partition, accountID,
 		Region:               region,
 		AutoscalingNamespace: namespace,
 		WAFScope:             c.WAFScope,
+		Backend:              c.Backend,
 	}
 }
 
@@ -169,6 +174,7 @@ func (c *Client) withPartitionAccountIDRegionAndScope(partition, accountID, regi
 		Region:               region,
 		AutoscalingNamespace: c.AutoscalingNamespace,
 		WAFScope:             scope,
+		Backend:              c.Backend,
 	}
 }
 
@@ -324,7 +330,7 @@ func configureAwsClient(ctx context.Context, logger zerolog.Logger, awsConfig *S
 	return awsCfg, err
 }
 
-func Configure(ctx context.Context, logger zerolog.Logger, spec specs.Source, _ source.Options) (schema.ClientMeta, error) {
+func Configure(ctx context.Context, logger zerolog.Logger, spec specs.Source, opts source.Options) (schema.ClientMeta, error) {
 	var awsConfig Spec
 	err := spec.UnmarshalSpec(&awsConfig)
 	if err != nil {
@@ -336,7 +342,8 @@ func Configure(ctx context.Context, logger zerolog.Logger, spec specs.Source, _
 		return nil, fmt.Errorf("spec validation failed: %w", err)
 	}
 
-	client := NewAwsClient(logger)
+	client := NewAwsClient(logger, opts.Backend)
+
 	var adminAccountSts AssumeRoleAPIClient
 
 	if awsConfig.Organization != nil {

diff --git a/plugins/source/aws/client/mocks/cloudtrail.go b/plugins/source/aws/client/mocks/cloudtrail.go
@@ -31,7 +31,7 @@ func AwsMockTestHelper(t *testing.T, table *schema.Table, builder func(*testing.
 		if err := spec.UnmarshalSpec(&awsSpec); err != nil {
 			return nil, fmt.Errorf("failed to unmarshal aws spec: %w", err)
 		}
-		c := NewAwsClient(l)
+		c := NewAwsClient(l, nil)
 		c.ServicesManager.InitServicesForPartitionAccountAndRegion("aws", "testAccount", "us-east-1", builder(t, ctrl))
 		c.Partition = "aws"
 		return &c, nil

@@ -233,6 +233,7 @@ func include(m reflect.Method) bool {
 	var exceptions = []string{
 		"QuerySchemaVersionMetadata",
 		"GenerateCredentialReport",
+		"LookupEvents",
 	}
 	if funk.ContainsString(exceptions, m.Name) {
 		return true

diff --git a/plugins/source/aws/docs/tables/README.md b/plugins/source/aws/docs/tables/README.md
diff --git a/plugins/source/aws/docs/tables/aws_cloudtrail_events.md b/plugins/source/aws/docs/tables/aws_cloudtrail_events.md
@@ -0,0 +1,26 @@
+# Table: aws_cloudtrail_events
+
+https://docs.aws.amazon.com/awscloudtrail/latest/APIReference/API_Event.html
+
+The primary key for this table is **event_id**.
+It supports incremental syncs based on the **event_time** column.
+
+## Columns
+
+| Name          | Type          |
+| ------------- | ------------- |
+|_cq_source_name|String|
+|_cq_sync_time|Timestamp|
+|_cq_id|UUID|
+|_cq_parent_id|UUID|
+|account_id|String|
+|region|String|
+|cloud_trail_event|JSON|
+|event_time (Incremental Key)|Timestamp|
+|access_key_id|String|
+|event_id (PK)|String|
+|event_name|String|
+|event_source|String|
+|read_only|String|
+|resources|JSON|
+|username|String|
@@ -146,6 +146,7 @@ func tables() []*schema.Table {
 		cloudfront.Distributions(),
 		cloudhsmv2.Backups(),
 		cloudhsmv2.Clusters(),
+		cloudtrail.Events(),
 		cloudtrail.Trails(),
 		cloudwatch.Alarms(),
 		cloudwatchlogs.LogGroups(),

@@ -0,0 +1,91 @@
+package cloudtrail
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/service/cloudtrail"
+	"github.com/aws/aws-sdk-go-v2/service/cloudtrail/types"
+	"github.com/cloudquery/cloudquery/plugins/source/aws/client"
+	"github.com/cloudquery/plugin-sdk/schema"
+	"github.com/cloudquery/plugin-sdk/transformers"
+)
+
+const tableName = "aws_cloudtrail_events"
+
+func Events() *schema.Table {
+	return &schema.Table{
+		Name:          tableName,
+		Description:   `https://docs.aws.amazon.com/awscloudtrail/latest/APIReference/API_Event.html`,
+		Resolver:      fetchCloudtrailEvents,
+		Multiplex:     client.ServiceAccountRegionMultiplexer("cloudtrail"),
+		Transform:     transformers.TransformWithStruct(&types.Event{}, transformers.WithPrimaryKeys("EventId")),
+		IsIncremental: true,
+		Columns: []schema.Column{
+			client.DefaultAccountIDColumn(false),
+			client.DefaultRegionColumn(false),
+			{
+				Name:     "cloud_trail_event",
+				Type:     schema.TypeJSON,
+				Resolver: schema.PathResolver("CloudTrailEvent"),
+			},
+			{
+				Name:     "event_time",
+				Type:     schema.TypeTimestamp,
+				Resolver: schema.PathResolver("EventTime"),
+				CreationOptions: schema.ColumnCreationOptions{
+					IncrementalKey: true,
+				},
+			},
+		},
+	}
+}
+
+func fetchCloudtrailEvents(ctx context.Context, meta schema.ClientMeta, parent *schema.Resource, res chan<- any) error {
+	cl := meta.(*client.Client)
+	svc := cl.Services().Cloudtrail
+
+	le := &cloudtrail.LookupEventsInput{}
+
+	if cl.Backend != nil {
+		value, err := cl.Backend.Get(ctx, tableName, cl.ID())
+		if err != nil {
+			return fmt.Errorf("failed to retrieve state from backend: %w", err)
+		}
+
+		if value != "" {
+			date, err := time.Parse(time.RFC3339Nano, value)
+			if err != nil {
+				return fmt.Errorf("retrieved invalid state value: %q %w", value, err)
+			}
+			le.StartTime = &date
+		}
+	}
+	var lastEventTime *time.Time
+	// var err error
+	paginator := cloudtrail.NewLookupEventsPaginator(svc, le)
+	for paginator.HasMorePages() {
+		page, err := paginator.NextPage(ctx)
+		if err != nil {
+			return err
+		}
+		res <- page.Events
+
+		// Retrieve the timestamp from the latest event
+		for _, event := range page.Events {
+			if lastEventTime == nil {
+				lastEventTime = event.EventTime
+				continue
+			}
+			if event.EventTime.After(*lastEventTime) {
+				lastEventTime = event.EventTime
+			}
+		}
+	}
+
+	if cl.Backend != nil && lastEventTime != nil {
+		return cl.Backend.Set(ctx, tableName, cl.ID(), lastEventTime.Format(time.RFC3339Nano))
+	}
+	return nil
+}
@@ -0,0 +1,39 @@
+package cloudtrail
+
+import (
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/cloudtrail"
+	"github.com/aws/aws-sdk-go-v2/service/cloudtrail/types"
+	"github.com/cloudquery/cloudquery/plugins/source/aws/client"
+	"github.com/cloudquery/cloudquery/plugins/source/aws/client/mocks"
+	"github.com/cloudquery/plugin-sdk/faker"
+	"github.com/golang/mock/gomock"
+)
+
+func buildCloudtrailEventsMock(t *testing.T, ctrl *gomock.Controller) client.Services {
+	m := mocks.NewMockCloudtrailClient(ctrl)
+	services := client.Services{
+		Cloudtrail: m,
+	}
+	event := types.Event{}
+	err := faker.FakeObject(&event)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	event.CloudTrailEvent = aws.String("{}")
+	m.EXPECT().LookupEvents(gomock.Any(), gomock.Any(), gomock.Any()).Return(
+		&cloudtrail.LookupEventsOutput{
+			Events: []types.Event{event},
+		},
+		nil,
+	)
+
+	return services
+}
+
+func TestCloudtrailEvents(t *testing.T) {
+	client.AwsMockTestHelper(t, Events(), buildCloudtrailEventsMock, client.TestOptions{})
+}
@@ -54,7 +54,7 @@ func testCredentialReportsWithNilValues(t *testing.T) {
 		ctx := context.Background()
 		ctrl := gomock.NewController(t)
 		services := buildCredentialReportsWithNilValues(ctrl)
-		cl := client.NewAwsClient(zerolog.Logger{})
+		cl := client.NewAwsClient(zerolog.Logger{}, nil)
 		cl.ServicesManager.InitServicesForPartitionAccountAndRegion("aws", "testAccount", "us-east-1", services)
 		cl.Partition = "aws"
 		cl.Region = "us-east-1"

@@ -39,6 +39,7 @@ spec:
   # unless otherwise indicated they are configuration parameters rather than configured resources
   skip_tables:
     - aws_ec2_vpc_endpoint_services # this resource includes services that are available from AWS as well as other AWS Accounts
+    - aws_cloudtrail_events
     - aws_docdb_cluster_parameter_groups
     - aws_docdb_engine_versions
     - aws_ec2_instance_types