Add HTTP basic auth support to the http_listener input (#3496)

etc/telegraf.conf

@@ -2862,6 +2862,11 @@
 #   ## Add service certificate and key
 #   tls_cert = "/etc/telegraf/cert.pem"
 #   tls_key = "/etc/telegraf/key.pem"
+#
+#   ## Optional username and password to accept for HTTP basic authentication.
+#   ## You probably want to make sure you have TLS configured above for this.
+#   basic_username = "foobar"
+#   basic_password = "barfoo"
 
 
 # # Read metrics from Kafka topic(s)

plugins/inputs/http_listener/README.md

@@ -12,6 +12,8 @@ Enable TLS by specifying the file names of a service TLS certificate and key.
 
 Enable mutually authenticated TLS and authorize client connections by signing certificate authority by including a list of allowed CA certificate file names in ````tls_allowed_cacerts````.
 
+Enable basic HTTP authentication of clients by specifying a username and password to check for. These credentials will be received from the client _as plain text_ if TLS is not configured.
+
 See: [Telegraf Input Data Formats](https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md#influx).
 
 **Example:**
@@ -39,4 +41,8 @@ This is a sample configuration for the plugin.
 
   ## MTLS
   tls_allowed_cacerts = ["/etc/telegraf/clientca.pem"]
+
+  ## Basic authentication
+  basic_username = "foobar"
+  basic_password = "barfoo"
 ```

plugins/inputs/http_listener/http_listener.go

@@ -3,6 +3,7 @@ package http_listener
 import (
 	"bytes"
 	"compress/gzip"
+	"crypto/subtle"
 	"crypto/tls"
 	"crypto/x509"
 	"io"
@@ -44,6 +45,9 @@ type HTTPListener struct {
 	TlsCert           string
 	TlsKey            string
 
+	BasicUsername string
+	BasicPassword string
+
 	mu sync.Mutex
 	wg sync.WaitGroup
 
@@ -64,6 +68,7 @@ type HTTPListener struct {
 	PingsRecv       selfstat.Stat
 	NotFoundsServed selfstat.Stat
 	BuffersCreated  selfstat.Stat
+	AuthFailures    selfstat.Stat
 }
 
 const sampleConfig = `
@@ -90,6 +95,11 @@ const sampleConfig = `
   ## Add service certificate and key
   tls_cert = "/etc/telegraf/cert.pem"
   tls_key = "/etc/telegraf/key.pem"
+
+  ## Optional username and password to accept for HTTP basic authentication.
+  ## You probably want to make sure you have TLS configured above for this.
+  # basic_username = "foobar"
+  # basic_password = "barfoo"
 `
 
 func (h *HTTPListener) SampleConfig() string {
@@ -124,6 +134,7 @@ func (h *HTTPListener) Start(acc telegraf.Accumulator) error {
 	h.PingsRecv = selfstat.Register("http_listener", "pings_received", tags)
 	h.NotFoundsServed = selfstat.Register("http_listener", "not_founds_served", tags)
 	h.BuffersCreated = selfstat.Register("http_listener", "buffers_created", tags)
+	h.AuthFailures = selfstat.Register("http_listener", "auth_failures", tags)
 
 	if h.MaxBodySize == 0 {
 		h.MaxBodySize = DEFAULT_MAX_BODY_SIZE
@@ -194,25 +205,29 @@ func (h *HTTPListener) ServeHTTP(res http.ResponseWriter, req *http.Request) {
 	case "/write":
 		h.WritesRecv.Incr(1)
 		defer h.WritesServed.Incr(1)
-		h.serveWrite(res, req)
+		h.AuthenticateIfSet(h.serveWrite, res, req)
 	case "/query":
 		h.QueriesRecv.Incr(1)
 		defer h.QueriesServed.Incr(1)
 		// Deliver a dummy response to the query endpoint, as some InfluxDB
 		// clients test endpoint availability with a query
+		h.AuthenticateIfSet(func(res http.ResponseWriter, req *http.Request) {
 			res.Header().Set("Content-Type", "application/json")
 			res.Header().Set("X-Influxdb-Version", "1.0")
 			res.WriteHeader(http.StatusOK)
 			res.Write([]byte("{\"results\":[]}"))
+		}, res, req)
 	case "/ping":
 		h.PingsRecv.Incr(1)
 		defer h.PingsServed.Incr(1)
 		// respond to ping requests
+		h.AuthenticateIfSet(func(res http.ResponseWriter, req *http.Request) {
 			res.WriteHeader(http.StatusNoContent)
+		}, res, req)
 	default:
 		defer h.NotFoundsServed.Incr(1)
 		// Don't know how to respond to calls to other endpoints
-		http.NotFound(res, req)
+		h.AuthenticateIfSet(http.NotFound, res, req)
 	}
 }
 
@@ -376,6 +391,23 @@ func (h *HTTPListener) getTLSConfig() *tls.Config {
 	return tlsConf
 }
 
+func (h *HTTPListener) AuthenticateIfSet(handler http.HandlerFunc, res http.ResponseWriter, req *http.Request) {
+	if h.BasicUsername != "" && h.BasicPassword != "" {
+		reqUsername, reqPassword, ok := req.BasicAuth()
+		if !ok ||
+			subtle.ConstantTimeCompare([]byte(reqUsername), []byte(h.BasicUsername)) != 1 ||
+			subtle.ConstantTimeCompare([]byte(reqPassword), []byte(h.BasicPassword)) != 1 {
+
+			h.AuthFailures.Incr(1)
+			http.Error(res, "Unauthorized.", http.StatusUnauthorized)
+			return
+		}
+		handler(res, req)
+	} else {
+		handler(res, req)
+	}
+}
+
 func init() {
 	inputs.Add("http_listener", func() telegraf.Input {
 		return &HTTPListener{

plugins/inputs/http_listener/http_listener_test.go

@@ -101,6 +101,9 @@ NsFlcGACj+/TvacFYlA6N2nyFeokzoqLX28Ddxdh2erXqJ4hYIhT1ik9tkLggs2z
 1T1084BquCuO6lIcOwJBALX4xChoMUF9k0IxSQzlz//seQYDkQNsE7y9IgAOXkzp
 RaR4pzgPbnKj7atG+2dBnffWfE+1Mcy0INDAO6WxPg0=
 -----END RSA PRIVATE KEY-----`
+
+	basicUsername = "test-username-please-ignore"
+	basicPassword = "super-secure-password!"
 )
 
 var (
@@ -120,6 +123,13 @@ func newTestHTTPListener() *HTTPListener {
 	return listener
 }
 
+func newTestHTTPAuthListener() *HTTPListener {
+	listener := newTestHTTPListener()
+	listener.BasicUsername = basicUsername
+	listener.BasicPassword = basicPassword
+	return listener
+}
+
 func newTestHTTPSListener() *HTTPListener {
 	initServiceCertFiles.Do(func() {
 		acaf, err := ioutil.TempFile("", "allowedCAFile.crt")
@@ -239,6 +249,24 @@ func TestWriteHTTPSWithClientAuth(t *testing.T) {
 	require.EqualValues(t, 204, resp.StatusCode)
 }
 
+func TestWriteHTTPBasicAuth(t *testing.T) {
+	listener := newTestHTTPAuthListener()
+
+	acc := &testutil.Accumulator{}
+	require.NoError(t, listener.Start(acc))
+	defer listener.Stop()
+
+	client := &http.Client{}
+
+	req, err := http.NewRequest("POST", createURL(listener, "http", "/write", "db=mydb"), bytes.NewBuffer([]byte(testMsg)))
+	require.NoError(t, err)
+	req.SetBasicAuth(basicUsername, basicPassword)
+	resp, err := client.Do(req)
+	require.NoError(t, err)
+	resp.Body.Close()
+	require.EqualValues(t, http.StatusNoContent, resp.StatusCode)
+}
+
 func TestWriteHTTP(t *testing.T) {
 	listener := newTestHTTPListener()