[8.14] Fix handling of custom Endpoint when using S3 + SQS

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

@@ -70,19 +70,29 @@ type s3Input struct {
 func newInput(config config, store beater.StateStore) (*s3Input, error) {
 	awsConfig, err := awscommon.InitializeAWSConfig(config.AWSConfig)
 
-	if config.AWSConfig.Endpoint != "" {
-		// Add a custom endpointResolver to the awsConfig so that all the requests are routed to this endpoint
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize AWS credentials: %w", err)
+	}
+
+	endpointUri, err := url.Parse(config.AWSConfig.Endpoint)
+
+	// A custom endpoint has been specified!
+	if err == nil && config.AWSConfig.Endpoint != "" && !strings.HasPrefix(endpointUri.Hostname(), "s3") {
+
+		// For backwards compat:
+		// If the endpoint does not start with S3, we will use the endpoint resolver to make all SDK requests use the specified endpoint
+		// If the endpoint does start with S3, we will use the default resolver uses the endpoint field but can replace s3 with the desired service name like sqs
+
 		awsConfig.EndpointResolverWithOptions = awssdk.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (awssdk.Endpoint, error) {
 			return awssdk.Endpoint{
-				PartitionID:   "aws",
-				URL:           config.AWSConfig.Endpoint,
-				SigningRegion: awsConfig.Region,
+				PartitionID:       "aws",
+				Source:            awssdk.EndpointSourceCustom,
+				URL:               config.AWSConfig.Endpoint,
+				SigningRegion:     awsConfig.Region,
+				HostnameImmutable: true,
 			}, nil
 		})
-	}
 
-	if err != nil {
-		return nil, fmt.Errorf("failed to initialize AWS credentials: %w", err)
 	}
 
 	return &s3Input{
@@ -112,16 +122,23 @@ func (in *s3Input) Run(inputContext v2.Context, pipeline beat.Pipeline) error {
 	defer cancelInputCtx()
 
 	if in.config.QueueURL != "" {
-		regionName, err := getRegionFromQueueURL(in.config.QueueURL, in.config.AWSConfig.Endpoint, in.config.RegionName)
-		if err != nil && in.config.RegionName == "" {
-			return fmt.Errorf("failed to get AWS region from queue_url: %w", err)
+		regionName, err := getRegionFromQueueURL(in.config.QueueURL, in.config.AWSConfig.Endpoint, in.config.AWSConfig.DefaultRegion)
+
+		// If we can't get a region from anywhere, error out
+		if err != nil && regionName == "" && in.config.RegionName == "" {
+			return fmt.Errorf("region not specified and failed to get AWS region from queue_url: %w", err)
 		}
 		var warn regionMismatchError
 		if errors.As(err, &warn) {
 			// Warn of mismatch, but go ahead with configured region name.
 			inputContext.Logger.Warnf("%v: using %q", err, regionName)
 		}
-		in.awsConfig.Region = regionName
+
+		// Ensure we don't overwrite region when getRegionFromURL fails
+		// Ensure we don't overwrite a user-specified region with a parsed region.
+		if regionName != "" && in.config.RegionName == "" {
+			in.awsConfig.Region = regionName
+		}
 
 		// Create SQS receiver and S3 notification processor.
 		receiver, err := in.createSQSReceiver(inputContext, pipeline)
@@ -322,17 +339,45 @@ var errBadQueueURL = errors.New("QueueURL is not in format: https://sqs.{REGION_
 
 func getRegionFromQueueURL(queueURL string, endpoint, defaultRegion string) (region string, err error) {
 	// get region from queueURL
+	// Example for custom domain queue: https://sqs.us-east-1.abc.xyz/12345678912/test-s3-logs
 	// Example for sqs queue: https://sqs.us-east-1.amazonaws.com/12345678912/test-s3-logs
 	// Example for vpce: https://vpce-test.sqs.us-east-1.vpce.amazonaws.com/12345678912/sqs-queue
 	u, err := url.Parse(queueURL)
 	if err != nil {
 		return "", fmt.Errorf(queueURL + " is not a valid URL")
 	}
+
+	e, err := url.Parse(endpoint)
+	if err != nil {
+		return "", fmt.Errorf(endpoint + " is not a valid URL")
+	}
+
 	if (u.Scheme == "https" || u.Scheme == "http") && u.Host != "" {
 		queueHostSplit := strings.SplitN(u.Host, ".", 3)
+		endpointSplit := strings.SplitN(e.Host, ".", 3)
 		// check for sqs queue url
+
+		// Parse a user-provided custom endpoint
+		if endpoint != "" && queueHostSplit[0] == "sqs" && len(queueHostSplit) == 3 && len(endpointSplit) == 3 {
+			// Check if everything after the second dot in the queue url matches everything after the second dot in the endpoint
+			endpointMatchesQueueUrl := strings.SplitN(u.Hostname(), ".", 3)[2] == strings.SplitN(e.Hostname(), ".", 3)[2]
+			if !endpointMatchesQueueUrl {
+				// We couldn't resolve the URL
+				// We cannot infer the region by matching the endpoint and queue url, return the default region with a region mismatch warning
+				return defaultRegion, regionMismatchError{queueURLRegion: queueHostSplit[1], defaultRegion: endpointSplit[1]}
+			}
+
+			region = queueHostSplit[1]
+			if defaultRegion != "" && region != defaultRegion {
+				return region, regionMismatchError{queueURLRegion: region, defaultRegion: defaultRegion}
+			}
+			return region, nil
+		}
+
+		// Parse a standard SQS url
 		if len(queueHostSplit) == 3 && queueHostSplit[0] == "sqs" {
-			if queueHostSplit[2] == endpoint || (endpoint == "" && strings.HasPrefix(queueHostSplit[2], "amazonaws.")) {
+			// handle endpoint with no scheme, handle endpoint with scheme
+			if queueHostSplit[2] == endpoint || queueHostSplit[2] == e.Host || (endpoint == "" && strings.HasPrefix(queueHostSplit[2], "amazonaws.")) {
 				region = queueHostSplit[1]
 				if defaultRegion != "" && region != defaultRegion {
 					return defaultRegion, regionMismatchError{queueURLRegion: region, defaultRegion: defaultRegion}

x-pack/filebeat/input/awss3/input_test.go

@@ -69,6 +69,18 @@ func TestGetRegionFromQueueURL(t *testing.T) {
 			endpoint: "abc.xyz",
 			want:     "us-east-1",
 		},
+		{
+			name:     "abc.xyz_and_domain_with_matching_endpoint_and_scheme",
+			queueURL: "https://sqs.us-east-1.abc.xyz/627959692251/test-s3-logs",
+			endpoint: "https://abc.xyz",
+			want:     "us-east-1",
+		},
+		{
+			name:     "abc.xyz_and_domain_with_matching_url_endpoint",
+			queueURL: "https://sqs.us-east-1.abc.xyz/627959692251/test-s3-logs",
+			endpoint: "https://s3.us-east-1.abc.xyz",
+			want:     "us-east-1",
+		},
 		{
 			name:     "abc.xyz_and_domain_with_blank_endpoint",
 			queueURL: "https://sqs.us-east-1.abc.xyz/627959692251/test-s3-logs",