From 898c78532c7cd0f15a95e630d09602573dce4f3e Mon Sep 17 00:00:00 2001 From: kaiyan-sheng Date: Mon, 23 Dec 2019 14:01:11 -0700 Subject: [PATCH 1/4] Check content type when reading s3 files --- x-pack/filebeat/input/s3/input.go | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/x-pack/filebeat/input/s3/input.go b/x-pack/filebeat/input/s3/input.go index 3aa130a11f9..3fe50175dd1 100644 --- a/x-pack/filebeat/input/s3/input.go +++ b/x-pack/filebeat/input/s3/input.go @@ -440,17 +440,19 @@ func (p *s3Input) newS3BucketReader(svc s3iface.ClientAPI, s3Info s3Info) (*bufi return nil, errors.New("s3 get object response body is empty") } - if strings.HasSuffix(s3Info.key, ".gz") { - gzipReader, err := gzip.NewReader(resp.Body) - + // Check content-type + switch *resp.ContentType { + case "text/plain", "text/plain; charset=utf-8": + return bufio.NewReader(resp.Body), nil + case "application/x-gzip": + reader, err := gzip.NewReader(resp.Body) if err != nil { return nil, errors.Wrapf(err, "Failed to decompress gzipped file %v", s3Info.key) } - - return bufio.NewReader(gzipReader), nil + return bufio.NewReader(reader), nil + default: + return bufio.NewReader(resp.Body), nil } - - return bufio.NewReader(resp.Body), nil } func (p *s3Input) forwardEvent(event beat.Event) error { From 63faf4e76c08e82802e66421fa7cf9de0fbb5f9c Mon Sep 17 00:00:00 2001 From: kaiyan-sheng Date: Thu, 2 Jan 2020 10:34:57 -0700 Subject: [PATCH 2/4] Check resp.ContentType and filename --- x-pack/filebeat/input/s3/input.go | 28 ++++++++++++++++++++-------- 1 file changed, 20 insertions(+), 8 deletions(-) diff --git a/x-pack/filebeat/input/s3/input.go b/x-pack/filebeat/input/s3/input.go index 3fe50175dd1..4a8bd7b43ff 100644 --- a/x-pack/filebeat/input/s3/input.go +++ b/x-pack/filebeat/input/s3/input.go @@ -441,18 +441,30 @@ func (p *s3Input) newS3BucketReader(svc s3iface.ClientAPI, s3Info s3Info) (*bufi } // Check content-type - switch *resp.ContentType { - case "text/plain", "text/plain; charset=utf-8": - return bufio.NewReader(resp.Body), nil - case "application/x-gzip": - reader, err := gzip.NewReader(resp.Body) + if resp.ContentType != nil { + switch *resp.ContentType { + case "text/plain", "text/plain; charset=utf-8": + return bufio.NewReader(resp.Body), nil + case "application/x-gzip": + reader, err := gzip.NewReader(resp.Body) + if err != nil { + return nil, errors.Wrapf(err, "Failed to decompress gzipped file %v", s3Info.key) + } + return bufio.NewReader(reader), nil + default: + return bufio.NewReader(resp.Body), nil + } + } + + // If there is no content-type, check file name instead. + if strings.HasSuffix(s3Info.key, ".gz") { + gzipReader, err := gzip.NewReader(resp.Body) if err != nil { return nil, errors.Wrapf(err, "Failed to decompress gzipped file %v", s3Info.key) } - return bufio.NewReader(reader), nil - default: - return bufio.NewReader(resp.Body), nil + return bufio.NewReader(gzipReader), nil } + return bufio.NewReader(resp.Body), nil } func (p *s3Input) forwardEvent(event beat.Event) error { From 2b233f727109adc072981644cdc8542490db495d Mon Sep 17 00:00:00 2001 From: kaiyan-sheng Date: Fri, 3 Jan 2020 08:24:04 -0700 Subject: [PATCH 3/4] Add changelog --- CHANGELOG.next.asciidoc | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc index 9bd4bd1dea3..4330aebbada 100644 --- a/CHANGELOG.next.asciidoc +++ b/CHANGELOG.next.asciidoc @@ -218,6 +218,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d - Update Logstash module's Grok patterns to support Logstash 7.4 logs. {pull}14743[14743] - Fix a problem in Filebeat input httpjson where interval is not used as time.Duration. {issue}14752[14752] {pull}14753[14753] - Fix SSL config in input.yml for Filebeat httpjson input in the MISP module. {pull}14767[14767] +- Check content-type when creating new reader in s3 input. {pull}15252[15252] {issue}15225[15225] *Heartbeat* From 7e319adcdb6c0fb5d1c7f41619dca7fb6ecb2921 Mon Sep 17 00:00:00 2001 From: kaiyan-sheng Date: Tue, 7 Jan 2020 10:42:17 -0700 Subject: [PATCH 4/4] Remove case "text/plain" to use default instead --- x-pack/filebeat/input/s3/input.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/x-pack/filebeat/input/s3/input.go b/x-pack/filebeat/input/s3/input.go index 4a8bd7b43ff..2d83f9e4a18 100644 --- a/x-pack/filebeat/input/s3/input.go +++ b/x-pack/filebeat/input/s3/input.go @@ -443,8 +443,6 @@ func (p *s3Input) newS3BucketReader(svc s3iface.ClientAPI, s3Info s3Info) (*bufi // Check content-type if resp.ContentType != nil { switch *resp.ContentType { - case "text/plain", "text/plain; charset=utf-8": - return bufio.NewReader(resp.Body), nil case "application/x-gzip": reader, err := gzip.NewReader(resp.Body) if err != nil {