http: Save the request or response body independently

This patch adds two new configuration options to the http protocol
plugin in packetbeat: `include_request_body_for` and
`include_response_body_for`.  They work in a similar way to
`include_body_for`, but apply only to the request or response bodies.

When `include_body_for` is also specified, the behavior is as follows:

  include_body_for: [A]
  include_request_body_for: [B]
  include_response_body_for: [C]

Requests bodies will be included if the content-type matches either A or
B, and response bodies if it is A or C.

CHANGELOG.asciidoc

@@ -311,6 +311,7 @@ https://github.com/elastic/beats/compare/v6.0.0-beta2...master[Check the HEAD di
 - HTTP parses successfully on empty status phrase. {issue}6176[6176]
 - HTTP parser supports broken status line. {pull}6631[6631]
 - Fix high memory usage on HTTP body if body is not published. {pull}6680[6680]
+- Allow to capture the HTTP request or response bodies independently. {pull}6784[6784]
 
 *Winlogbeat*
 

packetbeat/_meta/beat.reference.yml

@@ -179,9 +179,18 @@ packetbeat.protocols:
   #send_all_headers: false
 
   # The list of content types for which Packetbeat includes the full HTTP
-  # payload in the response field.
+  # payload. If the request's or response's Content-Type matches any on this
+  # list, the full body will be included under the request or response field.
   #include_body_for: []
 
+  # The list of content types for which Packetbeat includes the full HTTP
+  # request payload.
+  #include_request_body_for: []
+
+  # The list of content types for which Packetbeat includes the full HTTP
+  # response payload.
+  #include_response_body_for: []
+
   # If the Cookie or Set-Cookie headers are sent, this option controls whether
   # they are split into individual values.
   #split_cookie: false

packetbeat/packetbeat.reference.yml

@@ -179,9 +179,18 @@ packetbeat.protocols:
   #send_all_headers: false
 
   # The list of content types for which Packetbeat includes the full HTTP
-  # payload in the response field.
+  # payload. If the request's or response's Content-Type matches any on this
+  # list, the full body will be included under the request or response field.
   #include_body_for: []
 
+  # The list of content types for which Packetbeat includes the full HTTP
+  # request payload.
+  #include_request_body_for: []
+
+  # The list of content types for which Packetbeat includes the full HTTP
+  # response payload.
+  #include_response_body_for: []
+
   # If the Cookie or Set-Cookie headers are sent, this option controls whether
   # they are split into individual values.
   #split_cookie: false

packetbeat/protos/http/config.go

@@ -7,15 +7,17 @@ import (
 )
 
 type httpConfig struct {
-	config.ProtocolCommon `config:",inline"`
-	SendAllHeaders        bool     `config:"send_all_headers"`
-	SendHeaders           []string `config:"send_headers"`
-	SplitCookie           bool     `config:"split_cookie"`
-	RealIPHeader          string   `config:"real_ip_header"`
-	IncludeBodyFor        []string `config:"include_body_for"`
-	HideKeywords          []string `config:"hide_keywords"`
-	RedactAuthorization   bool     `config:"redact_authorization"`
-	MaxMessageSize        int      `config:"max_message_size"`
+	config.ProtocolCommon  `config:",inline"`
+	SendAllHeaders         bool     `config:"send_all_headers"`
+	SendHeaders            []string `config:"send_headers"`
+	SplitCookie            bool     `config:"split_cookie"`
+	RealIPHeader           string   `config:"real_ip_header"`
+	IncludeBodyFor         []string `config:"include_body_for"`
+	IncludeRequestBodyFor  []string `config:"include_request_body_for"`
+	IncludeResponseBodyFor []string `config:"include_response_body_for"`
+	HideKeywords           []string `config:"hide_keywords"`
+	RedactAuthorization    bool     `config:"redact_authorization"`
+	MaxMessageSize         int      `config:"max_message_size"`
 }
 
 var (

packetbeat/protos/http/http.go

@@ -122,7 +122,12 @@ func (http *httpPlugin) setFromConfig(config *httpConfig) {
 	http.splitCookie = config.SplitCookie
 	http.parserConfig.realIPHeader = strings.ToLower(config.RealIPHeader)
 	http.transactionTimeout = config.TransactionTimeout
-	http.parserConfig.includeBodyFor = config.IncludeBodyFor
+	for _, list := range [][]string{config.IncludeBodyFor, config.IncludeRequestBodyFor} {
+		http.parserConfig.includeRequestBodyFor = append(http.parserConfig.includeRequestBodyFor, list...)
+	}
+	for _, list := range [][]string{config.IncludeBodyFor, config.IncludeResponseBodyFor} {
+		http.parserConfig.includeResponseBodyFor = append(http.parserConfig.includeResponseBodyFor, list...)
+	}
 	http.maxMessageSize = config.MaxMessageSize
 
 	if config.SendAllHeaders {

packetbeat/protos/http/http_parser.go

@@ -67,11 +67,12 @@ type parser struct {
 }
 
 type parserConfig struct {
-	realIPHeader     string
-	sendHeaders      bool
-	sendAllHeaders   bool
-	headersWhitelist map[string]bool
-	includeBodyFor   []string
+	realIPHeader           string
+	sendHeaders            bool
+	sendAllHeaders         bool
+	headersWhitelist       map[string]bool
+	includeRequestBodyFor  []string
+	includeResponseBodyFor []string
 }
 
 var (
@@ -260,7 +261,11 @@ func (parser *parser) parseHeaders(s *stream, m *message) (cont, ok, complete bo
 			return false, true, true
 		}
 
-		m.sendBody = parser.shouldIncludeInBody(m.contentType)
+		if m.isRequest {
+			m.sendBody = parser.shouldIncludeInBody(m.contentType, parser.config.includeRequestBodyFor)
+		} else {
+			m.sendBody = parser.shouldIncludeInBody(m.contentType, parser.config.includeResponseBodyFor)
+		}
 		m.saveBody = m.sendBody || (m.contentLength > 0 && bytes.Contains(m.contentType, []byte("urlencoded")))
 
 		if m.isChunked {
@@ -531,8 +536,8 @@ func (*parser) parseBodyChunkedWaitFinalCRLF(s *stream, m *message) (ok, complet
 	return true, true
 }
 
-func (parser *parser) shouldIncludeInBody(contenttype []byte) bool {
-	for _, include := range parser.config.includeBodyFor {
+func (parser *parser) shouldIncludeInBody(contenttype []byte, capturedContentTypes []string) bool {
+	for _, include := range capturedContentTypes {
 		if bytes.Contains(contenttype, []byte(include)) {
 			if isDebug {
 				debugf("Should Include Body = true Content-Type %s include_body %s",

packetbeat/protos/http/http_test.go

@@ -1164,7 +1164,8 @@ func TestHttpParser_includeBodyFor(t *testing.T) {
 
 	var store eventStore
 	http := httpModForTests(&store)
-	http.parserConfig.includeBodyFor = []string{"application/x-foo", "text/plain"}
+	http.parserConfig.includeRequestBodyFor = []string{"application/x-foo", "text/plain"}
+	http.parserConfig.includeResponseBodyFor = []string{"application/x-foo", "text/plain"}
 
 	tcptuple := testCreateTCPTuple()
 	packet := protos.Packet{Payload: req}
@@ -1305,6 +1306,9 @@ func TestHttp_configsSettingAll(t *testing.T) {
 	config.SendAllHeaders = true
 	config.SplitCookie = true
 	config.RealIPHeader = "X-Forwarded-For"
+	config.IncludeBodyFor = []string{"body"}
+	config.IncludeRequestBodyFor = []string{"req1", "req2"}
+	config.IncludeResponseBodyFor = []string{"resp1", "resp2", "resp3"}
 
 	// Set config
 	http.setFromConfig(&config)
@@ -1320,6 +1324,8 @@ func TestHttp_configsSettingAll(t *testing.T) {
 	assert.True(t, http.parserConfig.sendAllHeaders)
 	assert.Equal(t, config.SplitCookie, http.splitCookie)
 	assert.Equal(t, strings.ToLower(config.RealIPHeader), http.parserConfig.realIPHeader)
+	assert.Equal(t, append(config.IncludeBodyFor, config.IncludeRequestBodyFor...), http.parserConfig.includeRequestBodyFor)
+	assert.Equal(t, append(config.IncludeBodyFor, config.IncludeResponseBodyFor...), http.parserConfig.includeResponseBodyFor)
 }
 
 func TestHttp_configsSettingHeaders(t *testing.T) {
@@ -1341,6 +1347,63 @@ func TestHttp_configsSettingHeaders(t *testing.T) {
 	}
 }
 
+func TestHttp_includeBodies(t *testing.T) {
+	reqTp := "PUT /node HTTP/1.1\r\n" +
+		"Host: server\r\n" +
+		"Content-Length: 12\r\n" +
+		"Content-Type: %s\r\n" +
+		"\r\n" +
+		"request_body"
+	respTp := "HTTP/1.1 200 OK\r\n" +
+		"Content-Length: 5\r\n" +
+		"Content-Type: %s\r\n" +
+		"\r\n" +
+		"done."
+	var store eventStore
+	http := httpModForTests(&store)
+	config := defaultConfig
+	config.IncludeBodyFor = []string{"both"}
+	config.IncludeRequestBodyFor = []string{"req1", "req2"}
+	config.IncludeResponseBodyFor = []string{"resp1", "resp2", "resp3"}
+	http.setFromConfig(&config)
+
+	tcptuple := testCreateTCPTuple()
+
+	for idx, testCase := range []struct {
+		requestCt, responseCt   string
+		hasRequest, hasResponse bool
+	}{
+		{"none", "none", false, false},
+		{"both", "other", true, false},
+		{"other", "both", false, true},
+		{"both", "both", true, true},
+		{"req1", "none", true, false},
+		{"none", "req1", false, false},
+		{"req2", "resp1", true, true},
+		{"none", "resp2", false, true},
+		{"resp3", "req2", false, false},
+	} {
+		msg := fmt.Sprintf("test case %d (%s, %s)", idx, testCase.requestCt, testCase.responseCt)
+		req := fmt.Sprintf(reqTp, testCase.requestCt)
+		resp := fmt.Sprintf(respTp, testCase.responseCt)
+
+		packet := protos.Packet{Payload: []byte(req)}
+		private := protos.ProtocolData(&httpConnectionData{})
+		private = http.Parse(&packet, tcptuple, 0, private)
+
+		packet.Payload = []byte(resp)
+		private = http.Parse(&packet, tcptuple, 1, private)
+		http.ReceivedFin(tcptuple, 1, private)
+
+		trans := expectTransaction(t, &store)
+		assert.NotNil(t, trans)
+		hasKey, _ := trans.HasKey("http.request.body")
+		assert.Equal(t, testCase.hasRequest, hasKey, msg)
+		hasKey, _ = trans.HasKey("http.response.body")
+		assert.Equal(t, testCase.hasResponse, hasKey, msg)
+	}
+}
+
 func benchmarkHTTPMessage(b *testing.B, data []byte) {
 	http := httpModForTests(nil)
 	parser := newParser(&http.parserConfig)