@@ -16,12 +16,12 @@ import (
1616 "time"
1717
1818 "github.com/aws/aws-lambda-go/events"
19- "github.com/aws/aws-sdk-go/aws"
20- "github.com/aws/aws-sdk-go/aws/session "
21- "github.com/aws/aws-sdk-go/service/s3"
22- "github.com/aws/aws-sdk-go/service /s3/s3iface "
23- "github.com/aws/aws-sdk-go/service/sqs "
24- "github.com/aws/aws-sdk-go/service/sqs/sqsiface "
19+ "github.com/aws/aws-sdk-go-v2 /aws"
20+ "github.com/aws/aws-sdk-go-v2/config "
21+ "github.com/aws/aws-sdk-go-v2 /service/s3"
22+ s3Manager "github.com/aws/aws-sdk-go-v2/feature /s3/manager "
23+ s3types "github.com/aws/aws-sdk-go-v2 /service/s3/types "
24+ "github.com/aws/aws-sdk-go-v2 /service/sqs"
2525 yaml "github.com/goccy/go-yaml"
2626 "github.com/prometheus/client_golang/prometheus"
2727 log "github.com/sirupsen/logrus"
@@ -32,6 +32,16 @@ import (
3232 "github.com/crowdsecurity/crowdsec/pkg/types"
3333)
3434
35+ type S3API interface {
36+ s3Manager.ListObjectsV2APIClient
37+ s3Manager.DownloadAPIClient
38+ }
39+
40+ type SQSAPI interface {
41+ ReceiveMessage (ctx context.Context , params * sqs.ReceiveMessageInput , optFns ... func (* sqs.Options )) (* sqs.ReceiveMessageOutput , error )
42+ DeleteMessage (ctx context.Context , params * sqs.DeleteMessageInput , optFns ... func (* sqs.Options )) (* sqs.DeleteMessageOutput , error )
43+ }
44+
3545type S3Configuration struct {
3646 configuration.DataSourceCommonCfg `yaml:",inline"`
3747 AwsProfile * string `yaml:"aws_profile"`
@@ -51,12 +61,12 @@ type S3Source struct {
5161 metricsLevel metrics.AcquisitionMetricsLevel
5262 Config S3Configuration
5363 logger * log.Entry
54- s3Client s3iface. S3API
55- sqsClient sqsiface. SQSAPI
64+ s3Client S3API
65+ sqsClient SQSAPI
5666 readerChan chan S3Object
5767 t * tomb.Tomb
5868 out chan types.Event
59- ctx aws .Context
69+ ctx context .Context
6070 cancel context.CancelFunc
6171}
6272
@@ -109,62 +119,60 @@ const (
109119)
110120
111121func (s * S3Source ) newS3Client () error {
112- options := session.Options {
113- SharedConfigState : session .SharedConfigEnable ,
114- }
115- if s .Config .AwsProfile != nil {
116- options .Profile = * s .Config .AwsProfile
122+ if s .s3Client != nil {
123+ return nil
117124 }
118125
119- sess , err := session . NewSessionWithOptions ( options )
120- if err != nil {
121- return fmt . Errorf ( "failed to create aws session: %w" , err )
126+ var loadOpts [] func ( * config. LoadOptions ) error
127+ if s . Config . AwsProfile != nil && * s . Config . AwsProfile != "" {
128+ loadOpts = append ( loadOpts , config . WithSharedConfigProfile ( * s . Config . AwsProfile ) )
122129 }
123-
124- config := aws .NewConfig ()
125- if s .Config .AwsRegion != "" {
126- config = config .WithRegion (s .Config .AwsRegion )
130+ region := s .Config .AwsRegion
131+ if region == "" {
132+ region = "us-east-1"
127133 }
128- if s .Config .AwsEndpoint != "" {
129- config = config .WithEndpoint (s .Config .AwsEndpoint )
134+ loadOpts = append (loadOpts , config .WithRegion (region ))
135+ loadOpts = append (loadOpts , config .WithCredentialsProvider (aws.AnonymousCredentials {}))
136+ cfg , err := config .LoadDefaultConfig (s .ctx , loadOpts ... )
137+ if err != nil {
138+ return fmt .Errorf ("failed to load aws config: %w" , err )
130139 }
131140
132- s . s3Client = s3 .New ( sess , config )
133- if s .s3Client == nil {
134- return errors . New ( "failed to create S3 client" )
141+ var clientOpts [] func ( * s3.Options )
142+ if s .Config . AwsEndpoint != "" {
143+ clientOpts = append ( clientOpts , func ( o * s3. Options ) { o . BaseEndpoint = aws . String ( s . Config . AwsEndpoint ) } )
135144 }
136145
146+ s .s3Client = s3 .NewFromConfig (cfg , clientOpts ... )
137147 return nil
138148}
139149
140150func (s * S3Source ) newSQSClient () error {
141- var sess * session.Session
142-
143- if s .Config .AwsProfile != nil {
144- sess = session .Must (session .NewSessionWithOptions (session.Options {
145- SharedConfigState : session .SharedConfigEnable ,
146- Profile : * s .Config .AwsProfile ,
147- }))
148- } else {
149- sess = session .Must (session .NewSessionWithOptions (session.Options {
150- SharedConfigState : session .SharedConfigEnable ,
151- }))
151+ if s .sqsClient != nil {
152+ return nil
152153 }
153154
154- if sess == nil {
155- return errors .New ("failed to create aws session" )
155+ var loadOpts []func (* config.LoadOptions ) error
156+ if s .Config .AwsProfile != nil && * s .Config .AwsProfile != "" {
157+ loadOpts = append (loadOpts , config .WithSharedConfigProfile (* s .Config .AwsProfile ))
156158 }
157- config := aws . NewConfig ()
158- if s . Config . AwsRegion ! ="" {
159- config = config . WithRegion ( s . Config . AwsRegion )
159+ region := s . Config . AwsRegion
160+ if region = ="" {
161+ region = "us-east-1"
160162 }
161- if s .Config .AwsEndpoint != "" {
162- config = config .WithEndpoint (s .Config .AwsEndpoint )
163+ loadOpts = append (loadOpts , config .WithRegion (region ))
164+ loadOpts = append (loadOpts , config .WithCredentialsProvider (aws.AnonymousCredentials {}))
165+ cfg , err := config .LoadDefaultConfig (s .ctx , loadOpts ... )
166+ if err != nil {
167+ return fmt .Errorf ("failed to load aws config: %w" , err )
163168 }
164- s .sqsClient = sqs .New (sess , config )
165- if s .sqsClient == nil {
166- return errors .New ("failed to create SQS client" )
169+
170+ var clientOpts []func (* sqs.Options )
171+ if s .Config .AwsEndpoint != "" {
172+ clientOpts = append (clientOpts , func (o * sqs.Options ) { o .BaseEndpoint = aws .String (s .Config .AwsEndpoint ) })
167173 }
174+
175+ s .sqsClient = sqs .NewFromConfig (cfg , clientOpts ... )
168176 return nil
169177}
170178
@@ -186,13 +194,13 @@ func (s *S3Source) readManager() {
186194 }
187195}
188196
189- func (s * S3Source ) getBucketContent () ([]* s3 .Object , error ) {
197+ func (s * S3Source ) getBucketContent () ([]s3types .Object , error ) {
190198 logger := s .logger .WithField ("method" , "getBucketContent" )
191199 logger .Debugf ("Getting bucket content for %s" , s .Config .BucketName )
192- bucketObjects := make ([]* s3 .Object , 0 )
200+ bucketObjects := make ([]s3types .Object , 0 )
193201 var continuationToken * string
194202 for {
195- out , err := s .s3Client .ListObjectsV2WithContext (s .ctx , & s3.ListObjectsV2Input {
203+ out , err := s .s3Client .ListObjectsV2 (s .ctx , & s3.ListObjectsV2Input {
196204 Bucket : aws .String (s .Config .BucketName ),
197205 Prefix : aws .String (s .Config .Prefix ),
198206 ContinuationToken : continuationToken ,
@@ -340,10 +348,10 @@ func (s *S3Source) sqsPoll() error {
340348 return nil
341349 default :
342350 logger .Trace ("Polling SQS queue" )
343- out , err := s .sqsClient .ReceiveMessageWithContext (s .ctx , & sqs.ReceiveMessageInput {
351+ out , err := s .sqsClient .ReceiveMessage (s .ctx , & sqs.ReceiveMessageInput {
344352 QueueUrl : aws .String (s .Config .SQSName ),
345- MaxNumberOfMessages : aws . Int64 ( 10 ) ,
346- WaitTimeSeconds : aws . Int64 ( 20 ) , // Probably no need to make it configurable ?
353+ MaxNumberOfMessages : 10 ,
354+ WaitTimeSeconds : 20 , // Probably no need to make it configurable ?
347355 })
348356 if err != nil {
349357 logger .Errorf ("Error while polling SQS: %s" , err )
@@ -359,21 +367,23 @@ func (s *S3Source) sqsPoll() error {
359367 if err != nil {
360368 logger .Errorf ("Error while parsing SQS message: %s" , err )
361369 // Always delete the message to avoid infinite loop
362- _ , err = s .sqsClient .DeleteMessage (& sqs.DeleteMessageInput {
363- QueueUrl : aws .String (s .Config .SQSName ),
364- ReceiptHandle : message .ReceiptHandle ,
365- })
370+ _ , err = s .sqsClient .DeleteMessage (s .ctx ,
371+ & sqs.DeleteMessageInput {
372+ QueueUrl : aws .String (s .Config .SQSName ),
373+ ReceiptHandle : message .ReceiptHandle ,
374+ })
366375 if err != nil {
367376 logger .Errorf ("Error while deleting SQS message: %s" , err )
368377 }
369378 continue
370379 }
371380 logger .Debugf ("Received SQS message for object %s/%s" , bucket , key )
372381 s .readerChan <- S3Object {Key : key , Bucket : bucket }
373- _ , err = s .sqsClient .DeleteMessage (& sqs.DeleteMessageInput {
374- QueueUrl : aws .String (s .Config .SQSName ),
375- ReceiptHandle : message .ReceiptHandle ,
376- })
382+ _ , err = s .sqsClient .DeleteMessage (s .ctx ,
383+ & sqs.DeleteMessageInput {
384+ QueueUrl : aws .String (s .Config .SQSName ),
385+ ReceiptHandle : message .ReceiptHandle ,
386+ })
377387 if err != nil {
378388 logger .Errorf ("Error while deleting SQS message: %s" , err )
379389 }
@@ -393,7 +403,7 @@ func (s *S3Source) readFile(bucket string, key string) error {
393403 "key" : key ,
394404 })
395405
396- output , err := s .s3Client .GetObjectWithContext (s .ctx , & s3.GetObjectInput {
406+ output , err := s .s3Client .GetObject (s .ctx , & s3.GetObjectInput {
397407 Bucket : aws .String (bucket ),
398408 Key : aws .String (key ),
399409 })
0 commit comments