[Filebeat][New Input] Http Input

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
+}