New Filebeat input: http_endpoint (#18298)

## What does this PR do?

This filebeat input configures a HTTP port listener, accepting JSON formatted POST requests, which again is formatted into a event, initially the event is created with the "json." prefix and expects the ingest pipeline to mutate the event during ingestion.

The initial set of features is based on the Logstash input plugin, but implemented differently: https://www.elastic.co/guide/en/logstash/current/plugins-inputs-http.html

## Why is it important?

This idea is based on a few different scenarios:

- The user already has a large beats installation and no Logstash, and do not want to install Logstash solely for a single feature.
- HTTP Input allows applications to be directly integrated with Elastic, without needing connectivity to Elasticsearch directly (or Logstash).
- Allows us to integrate and create modules for any product that supports HTTP POST events like SOAR, cloud applications, ticketing systems etc etc.

## Features currently implemented

- HTTP Basic Auth On/Off
- HTTP/HTTPS configurable
- Listening interface and port configurable
- Response code on success configurable
- Response body on success configurable
- Response header on success configurable
- Proper HTTP codes on both success and error responses
- Message prefix configurable
- URL to post to is configurable 
- SSL path to cert, key and CA is configurable.

CHANGELOG.next.asciidoc

@@ -341,6 +341,7 @@ field. You can revert this change by configuring tags for the module and omittin
 - Improve ECS categorization field mappings in system module. {issue}16031[16031] {pull}18065[18065]
 - Change the `json.*` input settings implementation to merge parsed json objects with existing objects in the event instead of fully replacing them. {pull}17958[17958]
 - Improve ECS categorization field mappings in osquery module. {issue}16176[16176] {pull}17881[17881]
+- Added http_endpoint input{pull}18298[18298]
 - Add support for array parsing in azure-eventhub input. {pull}18585[18585]
 - Added `observer.vendor`, `observer.product`, and `observer.type` to PANW module events. {pull}18223[18223]
 - The `logstash` module can now automatically detect the log file format (JSON or plaintext) and process it accordingly. {issue}9964[9964] {pull}18095[18095]

filebeat/docs/filebeat-options.asciidoc

@@ -47,6 +47,7 @@ You can configure {beatname_uc} to use the following inputs:
 * <<{beatname_lc}-input-container>>
 * <<{beatname_lc}-input-docker>>
 * <<{beatname_lc}-input-google-pubsub>>
+* <<{beatname_lc}-input-http_endpoint>>
 * <<{beatname_lc}-input-httpjson>>
 * <<{beatname_lc}-input-kafka>>
 * <<{beatname_lc}-input-log>>
@@ -73,6 +74,8 @@ include::inputs/input-docker.asciidoc[]
 
 include::../../x-pack/filebeat/docs/inputs/input-google-pubsub.asciidoc[]
 
+include::../../x-pack/filebeat/docs/inputs/input-http-endpoint.asciidoc[]
+
 include::../../x-pack/filebeat/docs/inputs/input-httpjson.asciidoc[]
 
 include::inputs/input-kafka.asciidoc[]

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

@@ -0,0 +1,116 @@
+[role="xpack"]
+
+:type: http_endpoint
+
+[id="{beatname_lc}-input-{type}"]
+=== HTTP Endpoint input
+
+++++
+<titleabbrev>HTTP Endpoint</titleabbrev>
+++++
+
+beta[]
+
+Use the `http_endpoint` input to create a HTTP listener that can receive incoming HTTP POST requests.
+
+This input can for example be used to receive incoming webhooks from a third-party application or service.
+
+Example configurations:
+
+Basic example:
+["source","yaml",subs="attributes"]
+----
+{beatname_lc}.inputs:
+- type: http_endpoint
+  enabled: true
+  listen_address: 192.168.1.1
+  listen_port: 8080
+----
+
+Custom response example:
+["source","yaml",subs="attributes"]
+----
+{beatname_lc}.inputs:
+- type: http_endpoint
+  enabled: true
+  listen_address: 192.168.1.1
+  listen_port: 8080
+  response_code: 200
+  response_body: '{"message": "success"}'
+  url: "/"
+  prefix: "json"
+----
+
+Basic auth and SSL example:
+["source","yaml",subs="attributes"]
+----
+{beatname_lc}.inputs:
+- type: http_endpoint
+  enabled: true
+  listen_address: 192.168.1.1
+  listen_port: 8080
+  ssl.enabled: true
+  ssl.certificate: "/home/user/server.pem"
+  ssl.key: "/home/user/server.key"
+  ssl.verification_mode: "none"
+  ssl.certificate_authority: "/home/user/ca.pem"
+  basic_auth: true
+  username: someuser
+  password: somepassword
+----
+
+
+==== Configuration options
+
+The `http_endpoint` input supports the following configuration options plus the
+<<{beatname_lc}-input-{type}-common-options>> described later.
+
+[float]
+==== `basic_auth`
+
+Enables or disables HTTP basic auth for each incoming request. If enabled then `username` and `password` will also need to be configured.
+
+[float]
+==== `username`
+
+If `basic_auth` is enabled, this is the username used for authentication against the HTTP listener. Requires `password` to also be set.
+
+[float]
+==== `password`
+
+If `basic_auth` is eanbled, this is the password used for authentication against the HTTP listener. Requires `username` to also be set.
+
+[float]
+==== `response_code`
+
+The HTTP response code returned upon success. Should be in the 2XX range.
+
+[float]
+==== `response_body`
+
+The response body returned upon success.
+
+[float]
+==== `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`.
+
+[float]
+==== `listen_port`
+
+Which port the listener binds to. Defaults to 8000
+
+[float]
+==== `url`
+
+This options specific which URL path to accept requests on. Defaults to `/`
+
+[float]
+==== `prefix`
+
+This option specifies which prefix the incoming request will be mapped to.
+
+[id="{beatname_lc}-input-{type}-common-options"]
+include::../../../../filebeat/docs/inputs/input-common-options.asciidoc[]
+
+:type!:

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

@@ -0,0 +1,48 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package http_endpoint
+
+import (
+	"encoding/json"
+	"errors"
+
+	"github.com/elastic/beats/v7/libbeat/common/transport/tlscommon"
+)
+
+// Config contains information about httpjson configuration
+type config struct {
+	TLS           *tlscommon.ServerConfig `config:"ssl"`
+	BasicAuth     bool                    `config:"basic_auth"`
+	Username      string                  `config:"username"`
+	Password      string                  `config:"password"`
+	ResponseCode  int                     `config:"response_code" validate:"positive"`
+	ResponseBody  string                  `config:"response_body"`
+	ListenAddress string                  `config:"listen_address"`
+	ListenPort    string                  `config:"listen_port"`
+	URL           string                  `config:"url"`
+	Prefix        string                  `config:"prefix"`
+}
+
+func defaultConfig() config {
+	return config{
+		BasicAuth:     false,
+		Username:      "",
+		Password:      "",
+		ResponseCode:  200,
+		ResponseBody:  `{"message": "success"}`,
+		ListenAddress: "127.0.0.1",
+		ListenPort:    "8000",
+		URL:           "/",
+		Prefix:        "json",
+	}
+}
+
+func (c *config) Validate() error {
+	if !json.Valid([]byte(c.ResponseBody)) {
+		return errors.New("response_body must be valid JSON")
+	}
+
+	return nil
+}

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

@@ -0,0 +1,77 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package http_endpoint
+
+import (
+	"context"
+	"net/http"
+	"time"
+
+	"github.com/elastic/beats/v7/libbeat/common/transport/tlscommon"
+	"github.com/elastic/beats/v7/libbeat/logp"
+)
+
+type HttpServer struct {
+	log    *logp.Logger
+	server *http.Server
+	ctx    context.Context
+	stop   context.CancelFunc
+}
+
+func (h *HttpServer) Start() {
+	go func() {
+		if h.server.TLSConfig != nil {
+			h.log.Infof("Starting HTTPS server on %s", h.server.Addr)
+			//certificate is already loaded. That's why the parameters are empty
+			err := h.server.ListenAndServeTLS("", "")
+			if err != nil && err != http.ErrServerClosed {
+				h.log.Fatalf("Unable to start HTTPS server due to error: %v", err)
+			}
+		} else {
+			h.log.Infof("Starting HTTP server on %s", h.server.Addr)
+			err := h.server.ListenAndServe()
+			if err != nil && err != http.ErrServerClosed {
+				h.log.Fatalf("Unable to start HTTP server due to error: %v", err)
+			}
+		}
+	}()
+}
+
+func (h *HttpServer) Stop() {
+	h.log.Info("Stopping HTTP server")
+	h.stop()
+	if err := h.server.Shutdown(h.ctx); err != nil {
+		h.log.Fatalf("Unable to stop HTTP server due to error: %v", err)
+	}
+}
+
+func createServer(in *HttpEndpoint) (*HttpServer, error) {
+	mux := http.NewServeMux()
+	responseHandler := http.HandlerFunc(in.apiResponse)
+	mux.Handle(in.config.URL, in.validateRequest(responseHandler))
+	server := &http.Server{
+		Addr:    in.config.ListenAddress + ":" + in.config.ListenPort,
+		Handler: mux,
+	}
+
+	tlsConfig, err := tlscommon.LoadTLSServerConfig(in.config.TLS)
+	if err != nil {
+		return nil, err
+	}
+
+	if tlsConfig != nil {
+		server.TLSConfig = tlsConfig.BuildModuleConfig(in.config.ListenAddress)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	h := &HttpServer{
+		ctx:  ctx,
+		stop: cancel,
+		log:  logp.NewLogger("http_server"),
+	}
+	h.server = server
+
+	return h, nil
+}