[Filebeat] Enable HMAC Signature Validation for http_endpoint input

CHANGELOG.next.asciidoc

@@ -852,6 +852,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add parsing for `haproxy.http.request.raw_request_line` field  {issue}25480[25480] {pull}25482[25482]
 - Mark `filestream` input beta. {pull}25560[25560]
 - Update PanOS module to parse Global Protect & User ID logs. {issue}24722[24722] {issue}24724[24724] {pull}24927[24927]
+- Add HMAC signature validation support for http_endpoint input. {pull}24918[24918]
 
 *Heartbeat*
 

x-pack/filebeat/docs/inputs/input-http-endpoint.asciidoc

@@ -82,6 +82,19 @@ Authentication or checking that a specific header includes a specific value
   secret.value: secretheadertoken
 ----
 
+Validate a HMAC signature from a specific header
+["source","yaml",subs="attributes"]
+----
+{beatname_lc}.inputs:
+- type: http_endpoint
+  enabled: true
+  listen_address: 192.168.1.1
+  listen_port: 8080
+  hmac.header: "X-Hub-Signature-256"
+  hmac.key: "password123"
+  hmac.type: "sha256"
+  hmac.prefix: "sha256="
+----
 
 ==== Configuration options
 
@@ -113,6 +126,26 @@ The header to check for a specific value specified by `secret.value`. Certain we
 
 The secret stored in the header name specified by `secret.header`. Certain webhooks provide the possibility to include a special header and secret to identify the source.
 
+[float]
+==== `hmac.header`
+
+The name of the header that contains the HMAC signature: `X-Dropbox-Signature`, `X-Hub-Signature-256`, etc.
+
+[float]
+==== `hmac.key`
+
+The secret key used to calculate the HMAC signature. Typically, the webhook sender provides this value.
+
+[float]
+==== `hmac.type`
+
+The hash algorithm to use for the HMAC comparison. At this time the only valid values are `sha256` or `sha1`.
+
+[float]
+==== `hmac.prefix`
+
+The prefix for the signature. Certain webhooks prefix the HMAC signature with a value, for example `sha256=`.
+
 [float]
 ==== `content_type`
 
@@ -137,7 +170,7 @@ If multiple interfaces is present the `listen_address` can be set to control whi
 [float]
 ==== `listen_port`
 
-Which port the listener binds to. Defaults to 8000
+Which port the listener binds to. Defaults to 8000.
 
 [float]
 ==== `url`

x-pack/filebeat/input/http_endpoint/config.go

@@ -26,6 +26,10 @@ type config struct {
 	ContentType   string                  `config:"content_type"`
 	SecretHeader  string                  `config:"secret.header"`
 	SecretValue   string                  `config:"secret.value"`
+	HMACHeader    string                  `config:"hmac.header"`
+	HMACKey       string                  `config:"hmac.key"`
+	HMACType      string                  `config:"hmac.type"`
+	HMACPrefix    string                  `config:"hmac.prefix"`
 }
 
 func defaultConfig() config {
@@ -42,6 +46,10 @@ func defaultConfig() config {
 		ContentType:   "application/json",
 		SecretHeader:  "",
 		SecretValue:   "",
+		HMACHeader:    "",
+		HMACKey:       "",
+		HMACType:      "",
+		HMACPrefix:    "",
 	}
 }
 
@@ -52,12 +60,20 @@ func (c *config) Validate() error {
 
 	if c.BasicAuth {
 		if c.Username == "" || c.Password == "" {
-			return errors.New("Username and password required when basicauth is enabled")
+			return errors.New("username and password required when basicauth is enabled")
 		}
 	}
 
 	if (c.SecretHeader != "" && c.SecretValue == "") || (c.SecretHeader == "" && c.SecretValue != "") {
-		return errors.New("Both secret.header and secret.value must be set")
+		return errors.New("both secret.header and secret.value must be set")
+	}
+
+	if (c.HMACHeader != "" && c.HMACKey == "") || (c.HMACHeader == "" && c.HMACKey != "") {
+		return errors.New("both hmac.header and hmac.key must be set")
+	}
+
+	if c.HMACType != "" && !(c.HMACType == "sha1" || c.HMACType == "sha256") {
+		return errors.New("hmac.type must be sha1 or sha256")
 	}
 
 	return nil

x-pack/filebeat/input/http_endpoint/handler.go

@@ -29,8 +29,10 @@ type httpHandler struct {
 	responseBody string
 }
 
-var errBodyEmpty = errors.New("Body cannot be empty")
-var errUnsupportedType = errors.New("Only JSON objects are accepted")
+var (
+	errBodyEmpty       = errors.New("body cannot be empty")
+	errUnsupportedType = errors.New("only JSON objects are accepted")
+)
 
 // Triggers if middleware validation returns successful
 func (h *httpHandler) apiResponse(w http.ResponseWriter, r *http.Request) {
@@ -75,7 +77,9 @@ func withValidator(v validator, handler http.HandlerFunc) http.HandlerFunc {
 func sendErrorResponse(w http.ResponseWriter, status int, err error) {
 	w.Header().Add("Content-Type", "application/json")
 	w.WriteHeader(status)
-	fmt.Fprintf(w, `{"message": %q}`, err.Error())
+	e := json.NewEncoder(w)
+	e.SetEscapeHTML(false)
+	e.Encode(common.MapStr{"message": err.Error()})
 }
 
 func httpReadJsonObject(body io.Reader) (obj common.MapStr, status int, err error) {
@@ -94,7 +98,7 @@ func httpReadJsonObject(body io.Reader) (obj common.MapStr, status int, err erro
 
 	obj = common.MapStr{}
 	if err := json.Unmarshal(contents, &obj); err != nil {
-		return nil, http.StatusBadRequest, fmt.Errorf("Malformed JSON body: %w", err)
+		return nil, http.StatusBadRequest, fmt.Errorf("malformed JSON body: %w", err)
 	}
 
 	return obj, 0, nil

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

@@ -90,6 +90,10 @@ func (e *httpEndpoint) Run(ctx v2.Context, publisher stateless.Publisher) error
 		contentType:  e.config.ContentType,
 		secretHeader: e.config.SecretHeader,
 		secretValue:  e.config.SecretValue,
+		hmacHeader:   e.config.HMACHeader,
+		hmacKey:      e.config.HMACKey,
+		hmacType:     e.config.HMACType,
+		hmacPrefix:   e.config.HMACPrefix,
 	}
 
 	handler := &httpHandler{
@@ -117,7 +121,7 @@ func (e *httpEndpoint) Run(ctx v2.Context, publisher stateless.Publisher) error
 	}
 
 	if err != nil && err != http.ErrServerClosed {
-		return fmt.Errorf("Unable to start server due to error: %w", err)
+		return fmt.Errorf("unable to start server due to error: %w", err)
 	}
 	return nil
 }

x-pack/filebeat/input/http_endpoint/validate.go

@@ -5,9 +5,17 @@
 package http_endpoint
 
 import (
+	"bytes"
+	"crypto/hmac"
+	"crypto/sha1"
+	"crypto/sha256"
+	"encoding/hex"
 	"errors"
 	"fmt"
+	"hash"
+	"io/ioutil"
 	"net/http"
+	"strings"
 )
 
 type validator interface {
@@ -23,10 +31,18 @@ type apiValidator struct {
 	contentType        string
 	secretHeader       string
 	secretValue        string
+	hmacHeader         string
+	hmacKey            string
+	hmacType           string
+	hmacPrefix         string
 }
 
-var errIncorrectUserOrPass = errors.New("Incorrect username or password")
-var errIncorrectHeaderSecret = errors.New("Incorrect header or header secret")
+var (
+	errIncorrectUserOrPass    = errors.New("incorrect username or password")
+	errIncorrectHeaderSecret  = errors.New("incorrect header or header secret")
+	errMissingHMACHeader      = errors.New("missing HMAC header")
+	errIncorrectHMACSignature = errors.New("invalid HMAC signature")
+)
 
 func (v *apiValidator) ValidateHeader(r *http.Request) (int, error) {
 	if v.basicAuth {
@@ -43,11 +59,53 @@ func (v *apiValidator) ValidateHeader(r *http.Request) (int, error) {
 	}
 
 	if v.method != "" && v.method != r.Method {
-		return http.StatusMethodNotAllowed, fmt.Errorf("Only %v requests supported", v.method)
+		return http.StatusMethodNotAllowed, fmt.Errorf("only %v requests are allowed", v.method)
 	}
 
 	if v.contentType != "" && r.Header.Get("Content-Type") != v.contentType {
-		return http.StatusUnsupportedMediaType, fmt.Errorf("Wrong Content-Type header, expecting %v", v.contentType)
+		return http.StatusUnsupportedMediaType, fmt.Errorf("wrong Content-Type header, expecting %v", v.contentType)
+	}
+
+	if v.hmacHeader != "" && v.hmacKey != "" && v.hmacType != "" {
+		// Read HMAC signature from HTTP header.
+		hmacHeaderValue := r.Header.Get(v.hmacHeader)
+		if v.hmacHeader == "" {
+			return http.StatusUnauthorized, errMissingHMACHeader
+		}
+		if v.hmacPrefix != "" {
+			hmacHeaderValue = strings.TrimPrefix(hmacHeaderValue, v.hmacPrefix)
+		}
+		signature, err := hex.DecodeString(hmacHeaderValue)
+		if err != nil {
+			return http.StatusUnauthorized, fmt.Errorf("invalid HMAC signature hex: %w", err)
+		}
+
+		// We need access to the request body to validate the signature, but we
+		// must leave the body intact for future processing.
+		buf, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			return http.StatusInternalServerError, fmt.Errorf("failed to read request body: %w", err)
+		}
+		// Set r.Body back to untouched original value.
+		r.Body = ioutil.NopCloser(bytes.NewBuffer(buf))
+
+		// Compute HMAC of raw body.
+		var mac hash.Hash
+		switch v.hmacType {
+		case "sha256":
+			mac = hmac.New(sha256.New, []byte(v.hmacKey))
+		case "sha1":
+			mac = hmac.New(sha1.New, []byte(v.hmacKey))
+		default:
+			// Upstream config validation prevents this from happening.
+			panic(fmt.Errorf("unhandled hmac.type %q", v.hmacType))
+		}
+		mac.Write(buf)
+		actualMAC := mac.Sum(nil)
+
+		if !hmac.Equal(signature, actualMAC) {
+			return http.StatusUnauthorized, errIncorrectHMACSignature
+		}
 	}
 
 	return 0, nil

x-pack/filebeat/tests/system/test_http_endpoint.py

@@ -1,6 +1,8 @@
 import jinja2
 import requests
 import sys
+import hmac
+import hashlib
 import os
 import json
 from filebeat import BaseTest
@@ -100,7 +102,7 @@ def test_http_endpoint_wrong_content_header(self):
         print("response:", r.status_code, r.text)
 
         assert r.status_code == 415
-        assert r.text == '{"message": "Wrong Content-Type header, expecting application/json"}'
+        assert r.json()['message'] == 'wrong Content-Type header, expecting application/json'
 
     def test_http_endpoint_missing_auth_value(self):
         """
@@ -113,7 +115,7 @@ def test_http_endpoint_missing_auth_value(self):
 """
         self.get_config(options)
         filebeat = self.start_beat()
-        self.wait_until(lambda: self.log_contains("Username and password required when basicauth is enabled"))
+        self.wait_until(lambda: self.log_contains("username and password required when basicauth is enabled"))
         filebeat.kill_and_wait()
 
     def test_http_endpoint_wrong_auth_value(self):
@@ -139,7 +141,7 @@ def test_http_endpoint_wrong_auth_value(self):
         print("response:", r.status_code, r.text)
 
         assert r.status_code == 401
-        assert r.text == '{"message": "Incorrect username or password"}'
+        assert r.json()['message'] == 'incorrect username or password'
 
     def test_http_endpoint_wrong_auth_header(self):
         """
@@ -163,7 +165,7 @@ def test_http_endpoint_wrong_auth_header(self):
         print("response:", r.status_code, r.text)
 
         assert r.status_code == 401
-        assert r.text == '{"message": "Incorrect header or header secret"}'
+        assert r.json()['message'] == 'incorrect header or header secret'
 
     def test_http_endpoint_correct_auth_header(self):
         """
@@ -189,6 +191,63 @@ def test_http_endpoint_correct_auth_header(self):
         assert output[0]["input.type"] == "http_endpoint"
         assert output[0]["json.{}".format(self.prefix)] == message
 
+    def test_http_endpoint_valid_hmac(self):
+        """
+        Test http_endpoint input with valid hmac signature.
+        """
+        options = """
+  hmac.header: "X-Hub-Signature-256"
+  hmac.key: "password123"
+  hmac.type: "sha256"
+  hmac.prefix: "sha256="
+"""
+        self.get_config(options)
+        filebeat = self.start_beat()
+        self.wait_until(lambda: self.log_contains("Starting HTTP server on {}:{}".format(self.host, self.port)))
+
+        message = "somerandommessage"
+        payload = {self.prefix: message}
+
+        h = hmac.new("password123".encode(), json.dumps(payload).encode(), hashlib.sha256)
+        print(h.hexdigest())
+        headers = {"Content-Type": "application/json", "X-Hub-Signature-256": "sha256=" + h.hexdigest()}
+        r = requests.post(self.url, headers=headers, data=json.dumps(payload))
+
+        filebeat.check_kill_and_wait()
+        output = self.read_output()
+
+        assert r.text == '{"message": "success"}'
+        assert output[0]["input.type"] == "http_endpoint"
+        assert output[0]["json.{}".format(self.prefix)] == message
+
+    def test_http_endpoint_invalid_hmac(self):
+        """
+        Test http_endpoint input with invalid hmac signature.
+        """
+        options = """
+  hmac.header: "X-Hub-Signature-256"
+  hmac.key: "password123"
+  hmac.type: "sha256"
+  hmac.prefix: "sha256="
+"""
+        self.get_config(options)
+        filebeat = self.start_beat()
+        self.wait_until(lambda: self.log_contains("Starting HTTP server on {}:{}".format(self.host, self.port)))
+
+        message = "somerandommessage"
+        payload = {self.prefix: message}
+
+        h = hmac.new("password321".encode(), json.dumps(payload).encode(), hashlib.sha256)
+        headers = {"Content-Type": "application/json", "X-Hub-Signature-256": "shad256=" + h.hexdigest()}
+        r = requests.post(self.url, headers=headers, data=json.dumps(payload))
+
+        filebeat.check_kill_and_wait()
+
+        print("response:", r.status_code, r.text)
+
+        assert r.status_code == 401
+        self.assertRegex(r.json()['message'], 'invalid HMAC signature')
+
     def test_http_endpoint_empty_body(self):
         """
         Test http_endpoint input with empty body.
@@ -205,7 +264,7 @@ def test_http_endpoint_empty_body(self):
         print("response:", r.status_code, r.text)
 
         assert r.status_code == 406
-        assert r.text == '{"message": "Body cannot be empty"}'
+        assert r.json()['message'] == 'body cannot be empty'
 
     def test_http_endpoint_malformed_json(self):
         """
@@ -224,7 +283,7 @@ def test_http_endpoint_malformed_json(self):
         print("response:", r.status_code, r.text)
 
         assert r.status_code == 400
-        assert r.text.startswith('{"message": "Malformed JSON body:')
+        self.assertRegex(r.json()['message'], 'malformed JSON body')
 
     def test_http_endpoint_get_request(self):
         """
@@ -243,4 +302,4 @@ def test_http_endpoint_get_request(self):
         print("response:", r.status_code, r.text)
 
         assert r.status_code == 405
-        assert r.text == '{"message": "Only POST requests supported"}'
+        assert r.json()['message'] == 'only POST requests are allowed'