[receiver/webhook] Option to add a required header (#24452)

**Description:** 
Adding a feature - Allow option of adding a required header for incoming
webhook requests. If header doesn't match, returns a 401.

**Link to tracking Issue:**
[<24270>](#24270)

.chloggen/webhook-require-header.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: webhookreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Add an optional config setting to set a required header that all incoming requests must provide"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [24270]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

receiver/webhookeventreceiver/README.md

@@ -27,6 +27,9 @@ The following settings are optional:
 * `health_path` (default: '/health_check'): Path available for checking receiver status
 * `read_timeout` (default: '500ms'): Maximum wait time while attempting to read a received event
 * `write_timeout` (default: '500ms'): Maximum wait time while attempting to write a response
+* `required_header` (optional):  
+    * `key` (required if `required_header` config option is set): Represents the key portion of the required header.
+    * `value` (required if `required_header` config option is set): Represents the value portion of the required header.
 
 Example:
 ```yaml
@@ -36,6 +39,9 @@ receivers:
         read_timeout: "500ms"
         path: "eventsource/receiver"
         health_path: "eventreceiver/healthcheck"
+        required_header:
+            key: "required-header-key"
+            value: "required-header-value"
 ```
 The full list of settings exposed for this receiver are documented [here](./config.go) with a detailed sample configuration [here](./testdata/config.yaml)
 

receiver/webhookeventreceiver/config.go

@@ -15,15 +15,22 @@ var (
 	errMissingEndpointFromConfig   = errors.New("missing receiver server endpoint from config")
 	errReadTimeoutExceedsMaxValue  = errors.New("The duration specified for read_timeout exceeds the maximum allowed value of 10s")
 	errWriteTimeoutExceedsMaxValue = errors.New("The duration specified for write_timeout exceeds the maximum allowed value of 10s")
+	errRequiredHeader              = errors.New("both key and value are required to assign a required_header")
 )
 
 // Config defines configuration for the Generic Webhook receiver.
 type Config struct {
 	confighttp.HTTPServerSettings `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct
-	ReadTimeout                   string                   `mapstructure:"read_timeout"`  // wait time for reading request headers in ms. Default is twenty seconds.
-	WriteTimeout                  string                   `mapstructure:"write_timeout"` // wait time for writing request response in ms. Default is twenty seconds.
-	Path                          string                   `mapstructure:"path"`          // path for data collection. Default is <host>:<port>/services/collector
-	HealthPath                    string                   `mapstructure:"health_path"`   // path for health check api. Default is /services/collector/health
+	ReadTimeout                   string                   `mapstructure:"read_timeout"`    // wait time for reading request headers in ms. Default is twenty seconds.
+	WriteTimeout                  string                   `mapstructure:"write_timeout"`   // wait time for writing request response in ms. Default is twenty seconds.
+	Path                          string                   `mapstructure:"path"`            // path for data collection. Default is <host>:<port>/services/collector
+	HealthPath                    string                   `mapstructure:"health_path"`     // path for health check api. Default is /services/collector/health
+	RequiredHeader                RequiredHeader           `mapstructure:"required_header"` // optional setting to set a required header for all requests to have
+}
+
+type RequiredHeader struct {
+	Key   string `mapstructure:"key"`
+	Value string `mapstructure:"value"`
 }
 
 func (cfg *Config) Validate() error {
@@ -59,5 +66,9 @@ func (cfg *Config) Validate() error {
 		}
 	}
 
+	if (cfg.RequiredHeader.Key != "" && cfg.RequiredHeader.Value == "") || (cfg.RequiredHeader.Value != "" && cfg.RequiredHeader.Key == "") {
+		errs = multierr.Append(errs, errRequiredHeader)
+	}
+
 	return errs
 }

receiver/webhookeventreceiver/config_test.go

@@ -24,6 +24,7 @@ func TestValidateConfig(t *testing.T) {
 	errs = multierr.Append(errs, errMissingEndpointFromConfig)
 	errs = multierr.Append(errs, errReadTimeoutExceedsMaxValue)
 	errs = multierr.Append(errs, errWriteTimeoutExceedsMaxValue)
+	errs = multierr.Append(errs, errRequiredHeader)
 
 	tests := []struct {
 		desc   string
@@ -59,6 +60,32 @@ func TestValidateConfig(t *testing.T) {
 				WriteTimeout: "14s",
 			},
 		},
+		{
+			desc:   "RequiredHeader does not contain both a key and a value",
+			expect: errRequiredHeader,
+			conf: Config{
+				HTTPServerSettings: confighttp.HTTPServerSettings{
+					Endpoint: "",
+				},
+				RequiredHeader: RequiredHeader{
+					Key:   "key-present",
+					Value: "",
+				},
+			},
+		},
+		{
+			desc:   "RequiredHeader does not contain both a key and a value",
+			expect: errRequiredHeader,
+			conf: Config{
+				HTTPServerSettings: confighttp.HTTPServerSettings{
+					Endpoint: "",
+				},
+				RequiredHeader: RequiredHeader{
+					Key:   "",
+					Value: "value-present",
+				},
+			},
+		},
 		{
 			desc:   "Multiple invalid configs",
 			expect: errs,
@@ -68,6 +95,10 @@ func TestValidateConfig(t *testing.T) {
 				},
 				WriteTimeout: "14s",
 				ReadTimeout:  "15s",
+				RequiredHeader: RequiredHeader{
+					Key:   "",
+					Value: "value-present",
+				},
 			},
 		},
 	}
@@ -99,6 +130,10 @@ func TestLoadConfig(t *testing.T) {
 		WriteTimeout: "500ms",
 		Path:         "some/path",
 		HealthPath:   "health/path",
+		RequiredHeader: RequiredHeader{
+			Key:   "key-present",
+			Value: "value-present",
+		},
 	}
 
 	// create expected config

receiver/webhookeventreceiver/receiver.go

@@ -25,11 +25,12 @@ import (
 )
 
 var (
-	errNilLogsConsumer      = errors.New("missing a logs consumer")
-	errMissingEndpoint      = errors.New("missing a receiver endpoint")
-	errInvalidRequestMethod = errors.New("invalid method. Valid method is POST")
-	errInvalidEncodingType  = errors.New("invalid encoding type")
-	errEmptyResponseBody    = errors.New("request body content length is zero")
+	errNilLogsConsumer       = errors.New("missing a logs consumer")
+	errMissingEndpoint       = errors.New("missing a receiver endpoint")
+	errInvalidRequestMethod  = errors.New("invalid method. Valid method is POST")
+	errInvalidEncodingType   = errors.New("invalid encoding type")
+	errEmptyResponseBody     = errors.New("request body content length is zero")
+	errMissingRequiredHeader = errors.New("request was missing required header or incorrect header value")
 )
 
 const healthyResponse = `{"text": "Webhookevent receiver is healthy"}`
@@ -153,6 +154,14 @@ func (er *eventReceiver) handleReq(w http.ResponseWriter, r *http.Request, _ htt
 		return
 	}
 
+	if er.cfg.RequiredHeader.Key != "" {
+		requiredHeaderValue := r.Header.Get(er.cfg.RequiredHeader.Key)
+		if requiredHeaderValue != er.cfg.RequiredHeader.Value {
+			er.failBadReq(ctx, w, http.StatusUnauthorized, errMissingRequiredHeader)
+			return
+		}
+	}
+
 	encoding := r.Header.Get("Content-Encoding")
 	// only support gzip if encoding header is set.
 	if encoding != "" && encoding != "gzip" {

receiver/webhookeventreceiver/receiver_test.go

@@ -48,6 +48,10 @@ func TestCreateNewLogReceiver(t *testing.T) {
 				WriteTimeout: "210",
 				Path:         "/event",
 				HealthPath:   "/health",
+				RequiredHeader: RequiredHeader{
+					Key:   "key-present",
+					Value: "value-present",
+				},
 			},
 			consumer: consumertest.NewNop(),
 		},
@@ -147,6 +151,10 @@ func TestHandleReq(t *testing.T) {
 func TestFailedReq(t *testing.T) {
 	cfg := createDefaultConfig().(*Config)
 	cfg.Endpoint = "localhost:0"
+	headerCfg := createDefaultConfig().(*Config)
+	headerCfg.Endpoint = "localhost:0"
+	headerCfg.RequiredHeader.Key = "key-present"
+	headerCfg.RequiredHeader.Value = "value-present"
 
 	tests := []struct {
 		desc   string
@@ -186,6 +194,16 @@ func TestFailedReq(t *testing.T) {
 			}(),
 			status: http.StatusBadRequest,
 		},
+		{
+			desc: "Invalid required header value",
+			cfg:  *headerCfg,
+			req: func() *http.Request {
+				req := httptest.NewRequest("POST", "http://localhost/events", strings.NewReader("test"))
+				req.Header.Set("key-present", "incorrect-value")
+				return req
+			}(),
+			status: http.StatusUnauthorized,
+		},
 	}
 	for _, test := range tests {
 		t.Run(test.desc, func(t *testing.T) {

receiver/webhookeventreceiver/testdata/config.yaml

@@ -5,3 +5,6 @@ webhookevent/valid_config:
   write_timeout: "500ms"
   path: "some/path"
   health_path: "health/path"
+  required_header:
+    key: key-present
+    value: value-present