Merge branch 'main' into event-duration-windows-zero

Author: mauri870

CHANGELOG.next.asciidoc

@@ -428,6 +428,8 @@ otherwise no tag is added. {issue}42208[42208] {pull}42403[42403]
 - Added support for websocket keep_alive heartbeat in the streaming input. {issue}42277[42277] {pull}44204[44204]
 - Add milliseconds to document timestamp from awscloudwatch Filebeat input {pull}44306[44306]
 - Add support to the Active Directory entity analytics provider for device entities. {pull}44309[44309]
+- Add support for OPTIONS request to HTTP Endpoint input. {issue}43930[43930] {pull}44387[44387]
+
 
 *Auditbeat*
 

docs/reference/filebeat/filebeat-input-http_endpoint.md

@@ -20,7 +20,7 @@ These are the possible response codes from the server.
 | HTTP Response Code | Name | Reason |
 | --- | --- | --- |
 | 200 | OK | Returned on success. |
-| 400 | Bad Request | Returned if JSON body decoding fails or if `wait_for_completion_timeout` query validation fails. |
+| 400 | Bad Request | Returned if JSON body decoding fails, if an OPTIONS request is made and `options_headers` has not been set in the config, or if `wait_for_completion_timeout` query validation fails. |
 | 401 | Unauthorized | Returned when basic auth, secret header, or HMAC validation fails. |
 | 405 | Method Not Allowed | Returned if methods other than POST are used. |
 | 406 | Not Acceptable | Returned if the POST request does not contain a body. |
@@ -57,6 +57,21 @@ filebeat.inputs:
   prefix: "json"
 ```
 
+OPTIONS request-aware example:
+
+```yaml
+filebeat.inputs:
+- type: http_endpoint
+  enabled: true
+  listen_address: 192.168.1.1
+  listen_port: 8080
+  response_code: 200
+  options_headers:
+    Custom-Options-Header: [custom-options-header-value]
+  url: "/"
+  prefix: "json"
+```
+
 Map request to root of document example:
 
 ```yaml
@@ -285,6 +300,16 @@ The HTTP response code returned upon success. Should be in the 2XX range.
 The response body returned upon success.
 
 
+### `options_headers` [_options_headers]
+
+A set of response headers to add to the response for OPTIONS requests. Headers with the same canonical MIME header name will be replaced with the values in this configuration.
+
+
+### `options_response_code` [_options_response_code]
+
+The HTTP response code to return for OPTIONS requests. Defaults to `200` (OK).
+
+
 ### `listen_address` [_listen_address]
 
 If multiple interfaces is present the `listen_address` can be set to control which IP address the listener binds to. Defaults to `127.0.0.1`.

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

@@ -32,6 +32,8 @@ type config struct {
 	Password              string                  `config:"password"`
 	ResponseCode          int                     `config:"response_code" validate:"positive"`
 	ResponseBody          string                  `config:"response_body"`
+	OptionsHeaders        http.Header             `config:"options_headers"`
+	OptionsStatus         int                     `config:"options_response_code"`
 	ListenAddress         string                  `config:"listen_address"`
 	ListenPort            string                  `config:"listen_port"`
 	URL                   string                  `config:"url" validate:"required"`
@@ -69,6 +71,7 @@ func defaultConfig() config {
 		BasicAuth:     false,
 		ResponseCode:  200,
 		ResponseBody:  `{"message": "success"}`,
+		OptionsStatus: 200,
 		RetryAfter:    10,
 		ListenAddress: "127.0.0.1",
 		ListenPort:    "8000",

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

@@ -13,6 +13,7 @@ import (
 	"io"
 	"net"
 	"net/http"
+	"net/textproto"
 	"net/url"
 	"reflect"
 	"sort"
@@ -84,13 +85,21 @@ type handler struct {
 
 func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	txID := h.nextTxID()
-	h.log.Debugw("request", "url", r.URL, "tx_id", txID)
+	h.log.Debugw("request", "url", r.URL, "method", r.Method, "tx_id", txID)
 	status, err := h.validator.validateRequest(r)
 	if err != nil {
 		h.sendAPIErrorResponse(txID, w, r, h.log, status, err)
 		return
 	}
 
+	if r.Method == http.MethodOptions {
+		for k, v := range h.validator.optionsHeaders {
+			w.Header()[textproto.CanonicalMIMEHeaderKey(k)] = v
+		}
+		w.WriteHeader(h.validator.optionsStatus)
+		return
+	}
+
 	wait, err := getTimeoutWait(r.URL, h.log)
 	if err != nil {
 		h.sendAPIErrorResponse(txID, w, r, h.log, http.StatusBadRequest, err)

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

@@ -257,6 +257,52 @@ func Test_apiResponse(t *testing.T) {
 			wantStatus:   http.StatusOK,
 			wantResponse: `{"message": "success"}`,
 		},
+		{
+			name: "options_with_headers",
+			conf: func() config {
+				c := defaultConfig()
+				c.OptionsHeaders = http.Header{
+					"optional-response-header": {"Optional-response-value"},
+				}
+				return c
+			}(),
+			request: func() *http.Request {
+				req := httptest.NewRequest(http.MethodOptions, "/", nil)
+				req.Header.Set("Content-Type", "application/json")
+				return req
+			}(),
+			events:       []mapstr.M{},
+			wantStatus:   http.StatusOK,
+			wantResponse: "",
+		},
+		{
+			name: "options_empty_headers",
+			conf: func() config {
+				c := defaultConfig()
+				c.OptionsHeaders = http.Header{}
+				return c
+			}(),
+			request: func() *http.Request {
+				req := httptest.NewRequest(http.MethodOptions, "/", nil)
+				req.Header.Set("Content-Type", "application/json")
+				return req
+			}(),
+			events:       []mapstr.M{},
+			wantStatus:   http.StatusOK,
+			wantResponse: "",
+		},
+		{
+			name: "options_no_header",
+			conf: defaultConfig(),
+			request: func() *http.Request {
+				req := httptest.NewRequest(http.MethodOptions, "/", nil)
+				req.Header.Set("Content-Type", "application/json")
+				return req
+			}(),
+			events:       []mapstr.M{},
+			wantStatus:   http.StatusBadRequest,
+			wantResponse: `{"message":"OPTIONS requests are only allowed with options_headers set"}`,
+		},
 		{
 			name:  "hmac_hex",
 			setup: func(t *testing.T) { testutils.SkipIfFIPSOnly(t, "test HMAC uses SHA-1.") },

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

@@ -335,18 +335,20 @@ func newHandler(ctx context.Context, c config, prg *program, pub func(beat.Event
 		publish: pub,
 		metrics: metrics,
 		validator: apiValidator{
-			basicAuth:    c.BasicAuth,
-			username:     c.Username,
-			password:     c.Password,
-			method:       c.Method,
-			contentType:  c.ContentType,
-			secretHeader: c.SecretHeader,
-			secretValue:  c.SecretValue,
-			hmacHeader:   c.HMACHeader,
-			hmacKey:      c.HMACKey,
-			hmacType:     c.HMACType,
-			hmacPrefix:   c.HMACPrefix,
-			maxBodySize:  -1,
+			basicAuth:      c.BasicAuth,
+			username:       c.Username,
+			password:       c.Password,
+			method:         c.Method,
+			contentType:    c.ContentType,
+			secretHeader:   c.SecretHeader,
+			secretValue:    c.SecretValue,
+			hmacHeader:     c.HMACHeader,
+			hmacKey:        c.HMACKey,
+			hmacType:       c.HMACType,
+			hmacPrefix:     c.HMACPrefix,
+			maxBodySize:    -1,
+			optionsHeaders: c.OptionsHeaders,
+			optionsStatus:  c.OptionsStatus,
 		},
 		maxInFlight:           c.MaxInFlight,
 		retryAfter:            c.RetryAfter,

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

@@ -116,6 +116,81 @@ var serverPoolTests = []struct {
 			{"json": mapstr.M{"c": int64(3)}},
 		},
 	},
+	{
+		name:   "options_with_headers",
+		method: http.MethodOptions,
+		cfgs: []*httpEndpoint{{
+			addr: "127.0.0.1:9001",
+			config: config{
+				ResponseCode:   http.StatusOK,
+				ResponseBody:   `{"message": "success"}`,
+				OptionsStatus:  http.StatusOK,
+				OptionsHeaders: http.Header{"option-header": {"options-header-value"}},
+				ListenAddress:  "127.0.0.1",
+				ListenPort:     "9001",
+				URL:            "/",
+				Prefix:         "json",
+				ContentType:    "application/json",
+			},
+		}},
+		events: []target{
+			{
+				url: "http://127.0.0.1:9001/", wantHeader: http.Header{
+					"Content-Length": {"0"},
+					"Option-Header":  {"options-header-value"},
+				},
+			},
+		},
+		wantStatus: http.StatusOK,
+	},
+	{
+		name:   "options_empty_headers",
+		method: http.MethodOptions,
+		cfgs: []*httpEndpoint{{
+			addr: "127.0.0.1:9001",
+			config: config{
+				ResponseCode:   http.StatusOK,
+				ResponseBody:   `{"message": "success"}`,
+				OptionsStatus:  http.StatusOK,
+				OptionsHeaders: http.Header{},
+				ListenAddress:  "127.0.0.1",
+				ListenPort:     "9001",
+				URL:            "/",
+				Prefix:         "json",
+				ContentType:    "application/json",
+			},
+		}},
+		events: []target{
+			{
+				url: "http://127.0.0.1:9001/", wantHeader: http.Header{
+					"Content-Length": {"0"},
+				},
+			},
+		},
+		wantStatus: http.StatusOK,
+	},
+	{
+		name:   "options_no_headers",
+		method: http.MethodOptions,
+		cfgs: []*httpEndpoint{{
+			addr: "127.0.0.1:9001",
+			config: config{
+				ResponseCode:   http.StatusOK,
+				ResponseBody:   `{"message": "success"}`,
+				OptionsStatus:  http.StatusOK,
+				OptionsHeaders: nil,
+				ListenAddress:  "127.0.0.1",
+				ListenPort:     "9001",
+				URL:            "/",
+				Prefix:         "json",
+				ContentType:    "application/json",
+			},
+		}},
+		events: []target{
+			{url: "http://127.0.0.1:9001/", wantBody: `{"message":"OPTIONS requests are only allowed with options_headers set"}` + "\n"},
+		},
+		wantStatus: http.StatusBadRequest,
+	},
 	{
 		name: "distinct_ports",
 		cfgs: []*httpEndpoint{

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

@@ -38,6 +38,9 @@ type apiValidator struct {
 	hmacType           string
 	hmacPrefix         string
 	maxBodySize        int64
+
+	optionsHeaders http.Header
+	optionsStatus  int
 }
 
 func (v *apiValidator) validateRequest(r *http.Request) (status int, err error) {
@@ -54,7 +57,10 @@ func (v *apiValidator) validateRequest(r *http.Request) (status int, err error)
 		}
 	}
 
-	if v.method != "" && v.method != r.Method {
+	if !v.isMethodOK(r.Method) {
+		if r.Method == http.MethodOptions {
+			return http.StatusBadRequest, errors.New("OPTIONS requests are only allowed with options_headers set")
+		}
 		return http.StatusMethodNotAllowed, fmt.Errorf("only %v requests are allowed", v.method)
 	}
 
@@ -109,6 +115,13 @@ func (v *apiValidator) validateRequest(r *http.Request) (status int, err error)
 	return http.StatusAccepted, nil
 }
 
+func (v *apiValidator) isMethodOK(m string) bool {
+	if m == http.MethodOptions {
+		return v.optionsHeaders != nil
+	}
+	return v.method == "" || m == v.method
+}
+
 // decoders is the priority-ordered set of decoders to use for HMAC header values.
 var decoders = [...]func(string) ([]byte, error){
 	hex.DecodeString,