Add endpoint into AWS config (#16263)

* Add endpoint into AWS config

* add EndpointResolver for aws config

* Update endpoints with region name

* replace endpoint_region with regions config option

* Add endpoint to filebeat s3 input and aws module

* Add endpoint to autodiscover aws ec2 and elb providers

* update changelog

* Add EnrichAWSConfigWithEndpoint function in libbeat/common for both FB and MB

CHANGELOG.asciidoc

@@ -128,6 +128,7 @@ processing events. (CVE-2019-17596) See https://www.elastic.co/community/securit
 - Libbeat HTTP's Server can listen to a unix socket using the `unix:///tmp/hello.sock` syntax. {pull}13655[13655]
 - Libbeat HTTP's Server can listen to a Windows named pipe using the `npipe:///hello` syntax. {pull}13655[13655]
 - Adding new `Enterprise` license type to the licenser. {issue}14246[14246]
+- Add endpoint config in AWS config to support using custom endpoint accessing AWS APIs. {issue}16245[16245] {pull}16263[16263]
 
 *Auditbeat*
 

filebeat/docs/modules/aws.asciidoc

@@ -33,18 +33,12 @@ Example config:
 - module: aws
   s3access:
     enabled: false
-
-    # AWS SQS queue url
     #var.queue_url: https://sqs.myregion.amazonaws.com/123456/myqueue
-
-    # Filename of AWS credential file
-    # If not set "$HOME/.aws/credentials" is used on Linux/Mac
-    # "%UserProfile%\.aws\credentials" is used on Windows
-    # var.shared_credential_file: /etc/filebeat/aws_credentials
-
-    # Profile name for aws credential
-    # If not set the default profile is used
-    # var.credential_profile_name: fb-aws
+    #var.shared_credential_file: /etc/filebeat/aws_credentials
+    #var.credential_profile_name: fb-aws
+    #var.visibility_timeout: 300s
+    #var.api_timeout: 120s
+    #var.endpoint: amazonaws.com
 
   elb:
     enabled: false
@@ -104,6 +98,19 @@ Filename of AWS credential file.
 
 AWS credential profile name.
 
+*`var.visibility_timeout`*::
+
+The duration that the received messages are hidden from ReceiveMessage request.
+Default to be 300 seconds.
+
+*`var.api_timeout`*::
+
+Maximum duration before AWS API request will be interrupted. Default to be 120 seconds.
+
+*`var.endpoint`*::
+
+Custom endpoint used to access AWS APIs.
+
 [float]
 === cloudtrail fileset
 

x-pack/filebeat/filebeat.reference.yml

@@ -119,6 +119,9 @@ filebeat.modules:
     # Default to be 120s
     #var.api_timeout: 120s
 
+    # Custom endpoint used to access AWS APIs
+    #var.endpoint: amazonaws.com
+
   elb:
     enabled: false
 
@@ -142,6 +145,9 @@ filebeat.modules:
     # Default to be 120s
     #var.api_timeout: 120s
 
+    # Custom endpoint used to access AWS APIs
+    #var.endpoint: amazonaws.com
+
   vpcflow:
     enabled: false
 
@@ -165,6 +171,9 @@ filebeat.modules:
     # Default to be 120s
     #var.api_timeout: 120s
 
+    # Custom endpoint used to access AWS APIs
+    #var.endpoint: amazonaws.com
+
   cloudtrail:
     enabled: false
 
@@ -188,6 +197,9 @@ filebeat.modules:
     # Default to be 120s
     #var.api_timeout: 120s
 
+    # Custom endpoint used to access AWS APIs
+    #var.endpoint: amazonaws.com
+
 #-------------------------------- Azure Module --------------------------------
 - module: azure
   # All logs

x-pack/filebeat/input/s3/input.go

@@ -181,8 +181,9 @@ func (p *s3Input) Run() {
 
 		awsConfig := p.awsConfig.Copy()
 		awsConfig.Region = regionName
-		svcSQS := sqs.New(awsConfig)
-		svcS3 := s3.New(awsConfig)
+
+		svcSQS := sqs.New(awscommon.EnrichAWSConfigWithEndpoint(p.config.AwsConfig.Endpoint, "sqs", regionName, awsConfig))
+		svcS3 := s3.New(awscommon.EnrichAWSConfigWithEndpoint(p.config.AwsConfig.Endpoint, "s3", regionName, awsConfig))
 
 		p.workerWg.Add(1)
 		go p.run(svcSQS, svcS3, visibilityTimeout)

x-pack/filebeat/module/aws/_meta/config.yml

@@ -22,6 +22,9 @@
     # Default to be 120s
     #var.api_timeout: 120s
 
+    # Custom endpoint used to access AWS APIs
+    #var.endpoint: amazonaws.com
+
   elb:
     enabled: false
 
@@ -45,6 +48,9 @@
     # Default to be 120s
     #var.api_timeout: 120s
 
+    # Custom endpoint used to access AWS APIs
+    #var.endpoint: amazonaws.com
+
   vpcflow:
     enabled: false
 
@@ -68,6 +74,9 @@
     # Default to be 120s
     #var.api_timeout: 120s
 
+    # Custom endpoint used to access AWS APIs
+    #var.endpoint: amazonaws.com
+
   cloudtrail:
     enabled: false
 
@@ -90,3 +99,6 @@
     # Maximum duration before AWS API request will be interrupted
     # Default to be 120s
     #var.api_timeout: 120s
+
+    # Custom endpoint used to access AWS APIs
+    #var.endpoint: amazonaws.com

x-pack/filebeat/module/aws/_meta/docs.asciidoc

@@ -28,18 +28,12 @@ Example config:
 - module: aws
   s3access:
     enabled: false
-
-    # AWS SQS queue url
     #var.queue_url: https://sqs.myregion.amazonaws.com/123456/myqueue
-
-    # Filename of AWS credential file
-    # If not set "$HOME/.aws/credentials" is used on Linux/Mac
-    # "%UserProfile%\.aws\credentials" is used on Windows
-    # var.shared_credential_file: /etc/filebeat/aws_credentials
-
-    # Profile name for aws credential
-    # If not set the default profile is used
-    # var.credential_profile_name: fb-aws
+    #var.shared_credential_file: /etc/filebeat/aws_credentials
+    #var.credential_profile_name: fb-aws
+    #var.visibility_timeout: 300s
+    #var.api_timeout: 120s
+    #var.endpoint: amazonaws.com
 
   elb:
     enabled: false
@@ -99,6 +93,19 @@ Filename of AWS credential file.
 
 AWS credential profile name.
 
+*`var.visibility_timeout`*::
+
+The duration that the received messages are hidden from ReceiveMessage request.
+Default to be 300 seconds.
+
+*`var.api_timeout`*::
+
+Maximum duration before AWS API request will be interrupted. Default to be 120 seconds.
+
+*`var.endpoint`*::
+
+Custom endpoint used to access AWS APIs.
+
 [float]
 === cloudtrail fileset
 

x-pack/filebeat/module/aws/cloudtrail/config/cloudtrail.yml

@@ -20,6 +20,10 @@ visibility_timeout: {{ .visibility_timeout }}
 api_timeout: {{ .api_timeout }}
 {{ end }}
 
+{{ if .endpoint }}
+endpoint: {{ .endpoint }}
+{{ end }}
+
 {{ else if eq .input "file" }}
 
 type: log

x-pack/filebeat/module/aws/cloudtrail/manifest.yml

@@ -7,6 +7,7 @@ var:
   - name: credential_profile_name
   - name: visibility_timeout
   - name: api_timeout
+  - name: endpoint
 
 ingest_pipeline: ingest/pipeline.yml
 input: config/cloudtrail.yml

x-pack/filebeat/module/aws/elb/config/s3.yml

@@ -16,3 +16,7 @@ visibility_timeout: {{ .visibility_timeout }}
 {{ if .api_timeout }}
 api_timeout: {{ .api_timeout }}
 {{ end }}
+
+{{ if .endpoint }}
+endpoint: {{ .endpoint }}
+{{ end }}

x-pack/filebeat/module/aws/elb/manifest.yml

@@ -7,6 +7,7 @@ var:
   - name: credential_profile_name
   - name: visibility_timeout
   - name: api_timeout
+  - name: endpoint
 
 ingest_pipeline: ingest/pipeline.yml
 input: config/{{.input}}.yml

x-pack/filebeat/module/aws/s3access/config/s3.yml

@@ -16,3 +16,7 @@ visibility_timeout: {{ .visibility_timeout }}
 {{ if .api_timeout }}
 api_timeout: {{ .api_timeout }}
 {{ end }}
+
+{{ if .endpoint }}
+endpoint: {{ .endpoint }}
+{{ end }}

x-pack/filebeat/module/aws/s3access/manifest.yml

@@ -7,6 +7,7 @@ var:
   - name: credential_profile_name
   - name: visibility_timeout
   - name: api_timeout
+  - name: endpoint
 
 ingest_pipeline: ingest/pipeline.yml
 input: config/{{.input}}.yml

x-pack/filebeat/module/aws/vpcflow/config/input.yml

@@ -19,6 +19,10 @@ visibility_timeout: {{ .visibility_timeout }}
 api_timeout: {{ .api_timeout }}
 {{ end }}
 
+{{ if .endpoint }}
+endpoint: {{ .endpoint }}
+{{ end }}
+
 {{ else if eq .input "file" }}
 
 type: log

x-pack/filebeat/module/aws/vpcflow/manifest.yml

@@ -7,6 +7,7 @@ var:
   - name: credential_profile_name
   - name: visibility_timeout
   - name: api_timeout
+  - name: endpoint
 
 ingest_pipeline: ingest/pipeline.yml
 input: config/input.yml

x-pack/filebeat/modules.d/aws.yml.disabled

@@ -25,6 +25,9 @@
     # Default to be 120s
     #var.api_timeout: 120s
 
+    # Custom endpoint used to access AWS APIs
+    #var.endpoint: amazonaws.com
+
   elb:
     enabled: false
 
@@ -48,6 +51,9 @@
     # Default to be 120s
     #var.api_timeout: 120s
 
+    # Custom endpoint used to access AWS APIs
+    #var.endpoint: amazonaws.com
+
   vpcflow:
     enabled: false
 
@@ -71,6 +77,9 @@
     # Default to be 120s
     #var.api_timeout: 120s
 
+    # Custom endpoint used to access AWS APIs
+    #var.endpoint: amazonaws.com
+
   cloudtrail:
     enabled: false
 
@@ -93,3 +102,6 @@
     # Maximum duration before AWS API request will be interrupted
     # Default to be 120s
     #var.api_timeout: 120s
+
+    # Custom endpoint used to access AWS APIs
+    #var.endpoint: amazonaws.com

x-pack/libbeat/autodiscover/providers/aws/ec2/provider.go

@@ -57,7 +57,9 @@ func AutodiscoverBuilder(bus bus.Bus, uuid uuid.UUID, c *common.Config) (autodis
 	if config.Regions == nil {
 		// set default region to make initial aws api call
 		awsCfg.Region = "us-west-1"
-		svcEC2 := ec2.New(awsCfg)
+		svcEC2 := ec2.New(awscommon.EnrichAWSConfigWithEndpoint(
+			config.AWSConfig.Endpoint, "ec2", awsCfg.Region, awsCfg))
+
 		completeRegionsList, err := awsauto.GetRegions(svcEC2)
 		if err != nil {
 			return nil, err
@@ -72,7 +74,8 @@ func AutodiscoverBuilder(bus bus.Bus, uuid uuid.UUID, c *common.Config) (autodis
 			logp.Error(errors.Wrap(err, "error loading AWS config for aws_ec2 autodiscover provider"))
 		}
 		awsCfg.Region = region
-		clients = append(clients, ec2.New(awsCfg))
+		clients = append(clients, ec2.New(awscommon.EnrichAWSConfigWithEndpoint(
+			config.AWSConfig.Endpoint, "ec2", region, awsCfg)))
 	}
 
 	return internalBuilder(uuid, bus, config, newAPIFetcher(clients))

x-pack/libbeat/autodiscover/providers/aws/elb/provider.go

@@ -58,7 +58,9 @@ func AutodiscoverBuilder(bus bus.Bus, uuid uuid.UUID, c *common.Config) (autodis
 	if config.Regions == nil {
 		// set default region to make initial aws api call
 		awsCfg.Region = "us-west-1"
-		svcEC2 := ec2.New(awsCfg)
+		svcEC2 := ec2.New(awscommon.EnrichAWSConfigWithEndpoint(
+			config.AWSConfig.Endpoint, "ec2", awsCfg.Region, awsCfg))
+
 		completeRegionsList, err := awsauto.GetRegions(svcEC2)
 		if err != nil {
 			return nil, err
@@ -79,7 +81,8 @@ func AutodiscoverBuilder(bus bus.Bus, uuid uuid.UUID, c *common.Config) (autodis
 			logp.Err("error loading AWS config for aws_elb autodiscover provider: %s", err)
 		}
 		awsCfg.Region = region
-		clients = append(clients, elasticloadbalancingv2.New(awsCfg))
+		clients = append(clients, elasticloadbalancingv2.New(awscommon.EnrichAWSConfigWithEndpoint(
+			config.AWSConfig.Endpoint, "elasticloadbalancing", region, awsCfg)))
 	}
 
 	return internalBuilder(uuid, bus, config, newAPIFetcher(clients))

x-pack/libbeat/common/aws/credentials.go

@@ -17,6 +17,7 @@ type ConfigAWS struct {
 	SessionToken         string `config:"session_token"`
 	ProfileName          string `config:"credential_profile_name"`
 	SharedCredentialFile string `config:"shared_credential_file"`
+	Endpoint             string `config:"endpoint"`
 }
 
 // GetAWSCredentials function gets aws credentials from the config.
@@ -57,5 +58,23 @@ func GetAWSCredentials(config ConfigAWS) (awssdk.Config, error) {
 	if config.SharedCredentialFile != "" {
 		options = append(options, external.WithSharedConfigFiles([]string{config.SharedCredentialFile}))
 	}
-	return external.LoadDefaultAWSConfig(options...)
+
+	awsConfig, err := external.LoadDefaultAWSConfig(options...)
+	if err != nil {
+		return awsConfig, err
+	}
+	return awsConfig, nil
+}
+
+// EnrichAWSConfigWithEndpoint function enabled endpoint resolver for AWS
+// service clients when endpoint is given in config.
+func EnrichAWSConfigWithEndpoint(endpoint string, serviceName string, regionName string, awsConfig awssdk.Config) awssdk.Config {
+	if endpoint != "" {
+		if regionName == "" {
+			awsConfig.EndpointResolver = awssdk.ResolveWithEndpointURL("https://" + serviceName + "." + endpoint)
+		} else {
+			awsConfig.EndpointResolver = awssdk.ResolveWithEndpointURL("https://" + serviceName + "." + regionName + "." + endpoint)
+		}
+	}
+	return awsConfig
 }