[Filebeat] Fix hardcoded amazonaws.com endpoint

CHANGELOG.next.asciidoc

@@ -383,6 +383,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Improve Cisco ASA/FTD parsing of messages - better support for identity FW messages. Change network.bytes, source.bytes, and destination.bytes to long from integer since value can exceed integer capacity.  Add descriptions for various processors for easier pipeline editing in Kibana UI. {pull}23766[23766]
 - Updating Oauth2 flow for m365_defender fileset. {pull}24829[24829]
 - Improve PanOS parsing and ingest pipeline. {issue}22413[22413] {issue}22748[22748] {pull}24799[24799]
+- Fix S3 input validation for non amazonaws.com domains. {issue}24420[24420] {pull}24861[24861]
 
 *Heartbeat*
 

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

@@ -226,14 +226,20 @@ func (c *s3Collector) changeVisibilityTimeout(queueURL string, visibilityTimeout
 	return err
 }
 
-func getRegionFromQueueURL(queueURL string) (string, error) {
+func getRegionFromQueueURL(queueURL string, endpoint string) (string, error) {
 	// get region from queueURL
 	// Example: https://sqs.us-east-1.amazonaws.com/627959692251/test-s3-logs
-	queueURLSplit := strings.Split(queueURL, ".")
-	if queueURLSplit[0] == "https://sqs" && queueURLSplit[2] == "amazonaws" {
-		return queueURLSplit[1], nil
+	url, err := url.Parse(queueURL)
+	if err != nil {
+		return "", fmt.Errorf("QueueURL is not a valid URL")
+	}
+	if url.Scheme == "https" && url.Host != "" {
+		queueHostSplit := strings.Split(url.Host, ".")
+		if endpoint != "" && len(queueHostSplit) > 2 && (strings.Join(queueHostSplit[2:], ".") == endpoint || queueHostSplit[2] == "amazonaws") {
+			return queueHostSplit[1], nil
+		}
 	}
-	return "", fmt.Errorf("queueURL is not in format: https://sqs.{REGION_ENDPOINT}.amazonaws.com/{ACCOUNT_NUMBER}/{QUEUE_NAME}")
+	return "", fmt.Errorf("QueueURL is not in format: https://sqs.{REGION_ENDPOINT}.{ENDPOINT}/{ACCOUNT_NUMBER}/{QUEUE_NAME}")
 }
 
 // handle message

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

@@ -58,11 +58,25 @@ func (m *MockS3Client) GetObjectRequest(input *s3.GetObjectInput) s3.GetObjectRe
 
 func TestGetRegionFromQueueURL(t *testing.T) {
 	queueURL := "https://sqs.us-east-1.amazonaws.com/627959692251/test-s3-logs"
-	regionName, err := getRegionFromQueueURL(queueURL)
+	regionName, err := getRegionFromQueueURL(queueURL, "")
 	assert.NoError(t, err)
 	assert.Equal(t, "us-east-1", regionName)
 }
 
+func TestGetRegionFromQueueURLOtherEndpoint(t *testing.T) {
+	queueURL := "https://sqs.us-east-1.abc.xyz/627959692251/test-s3-logs"
+	regionName, err := getRegionFromQueueURL(queueURL, "abc.xyz")
+	assert.NoError(t, err)
+	assert.Equal(t, "us-east-1", regionName)
+}
+
+func TestGetRegionFromQueueURLBadEndpoint(t *testing.T) {
+	queueURL := "https://sqs.us-east-1.abc.xyz/627959692251/test-s3-logs"
+	regionName, err := getRegionFromQueueURL(queueURL, "")
+	assert.Error(t, err)
+	assert.Equal(t, "", regionName)
+}
+
 func TestHandleMessage(t *testing.T) {
 	casesPositive := []struct {
 		title           string

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

@@ -87,7 +87,7 @@ func (in *s3Input) createCollector(ctx v2.Context, pipeline beat.Pipeline) (*s3C
 		return nil, err
 	}
 
-	regionName, err := getRegionFromQueueURL(in.config.QueueURL)
+	regionName, err := getRegionFromQueueURL(in.config.QueueURL, in.config.AwsConfig.Endpoint)
 	if err != nil {
 		err := fmt.Errorf("getRegionFromQueueURL failed: %w", err)
 		log.Error(err)