TLS and MTLS enhancements to HTTPListener input plugin

plugins/inputs/http_listener/README.md

@@ -8,6 +8,10 @@ The `/write` endpoint supports the `precision` query parameter and can be set to
 
 When chaining Telegraf instances using this plugin, CREATE DATABASE requests receive a 200 OK response with message body `{"results":[]}` but they are not relayed. The output configuration of the Telegraf instance which ultimately submits data to InfluxDB determines the destination database.
 
+Enable TLS by specifying the file names of a service TLS certificate and key. Include one or more CA certificate file names if using a private PKI. 
+
+Enable mutually authenticated TLS and authorize client connections by signing certificate authority by including a list of allowed CA certificate file names in ````ssl_allowed_client_certificate_authorities````.
+
 See: [Telegraf Input Data Formats](https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md#influx).
 
 **Example:**
@@ -28,4 +32,12 @@ This is a sample configuration for the plugin.
   ## timeouts
   read_timeout = "10s"
   write_timeout = "10s"
+
+  ## HTTPS
+  ssl_certificate_authorities = ["/etc/ca.crt"]
+  ssl_certificate = "/etc/service.crt"
+  ssl_key = "/etc/service.key"
+
+  ## MTLS
+  ssl_allowed_client_certificate_authorities = ["/etc/ca.crt"]
 ```

plugins/inputs/http_listener/http_listener.go

@@ -3,7 +3,10 @@ package http_listener
 import (
 	"bytes"
 	"compress/gzip"
+	"crypto/tls"
+	"crypto/x509"
 	"io"
+	"io/ioutil"
 	"log"
 	"net"
 	"net/http"
@@ -37,6 +40,11 @@ type HTTPListener struct {
 	MaxLineSize    int
 	Port           int
 
+	SslAllowedClientCertificateAuthorities []string
+	SslCertificateAuthorities              []string
+	SslCertificate                         string
+	SslKey                                 string
+
 	mu sync.Mutex
 	wg sync.WaitGroup
 
@@ -75,6 +83,18 @@ const sampleConfig = `
   ## Maximum line size allowed to be sent in bytes.
   ## 0 means to use the default of 65536 bytes (64 kibibytes)
   max_line_size = 0
+
+  ## Set one or more allowed client CA certificate file names to 
+  ## enable mutually authenticated TLS connections
+  ssl_allowed_client_certificate_authorities = ["/etc/ca.crt"]
+
+  ## Add non-public root of trust certificate authorities
+  ssl_certificate_authorities = ["/etc/ca.crt"]
+
+  ## Add service certificate and key
+  ssl_certificate = "/etc/service.crt"
+  ssl_key = "/etc/service.key"
+
 `
 
 func (h *HTTPListener) SampleConfig() string {
@@ -117,10 +137,33 @@ func (h *HTTPListener) Start(acc telegraf.Accumulator) error {
 		h.MaxLineSize = DEFAULT_MAX_LINE_SIZE
 	}
 
+	if h.ReadTimeout.Duration < time.Second {
+		h.ReadTimeout.Duration = time.Second * 10
+	}
+	if h.WriteTimeout.Duration < time.Second {
+		h.WriteTimeout.Duration = time.Second * 10
+	}
+
 	h.acc = acc
 	h.pool = NewPool(200, h.MaxLineSize)
 
-	var listener, err = net.Listen("tcp", h.ServiceAddress)
+	tlsConf := h.getTLSConfig()
+
+	server := &http.Server{
+		Addr:         h.ServiceAddress,
+		Handler:      h,
+		ReadTimeout:  h.ReadTimeout.Duration,
+		WriteTimeout: h.WriteTimeout.Duration,
+		TLSConfig:    tlsConf,
+	}
+
+	var err error
+	var listener net.Listener
+	if tlsConf != nil {
+		listener, err = tls.Listen("tcp", h.ServiceAddress, tlsConf)
+	} else {
+		listener, err = net.Listen("tcp", h.ServiceAddress)
+	}
 	if err != nil {
 		return err
 	}
@@ -130,7 +173,7 @@ func (h *HTTPListener) Start(acc telegraf.Accumulator) error {
 	h.wg.Add(1)
 	go func() {
 		defer h.wg.Done()
-		h.httpListen()
+		server.Serve(h.listener)
 	}()
 
 	log.Printf("I! Started HTTP listener service on %s\n", h.ServiceAddress)
@@ -149,27 +192,6 @@ func (h *HTTPListener) Stop() {
 	log.Println("I! Stopped HTTP listener service on ", h.ServiceAddress)
 }
 
-// httpListen sets up an http.Server and calls server.Serve.
-// like server.Serve, httpListen will always return a non-nil error, for this
-// reason, the error returned should probably be ignored.
-// see https://golang.org/pkg/net/http/#Server.Serve
-func (h *HTTPListener) httpListen() error {
-	if h.ReadTimeout.Duration < time.Second {
-		h.ReadTimeout.Duration = time.Second * 10
-	}
-	if h.WriteTimeout.Duration < time.Second {
-		h.WriteTimeout.Duration = time.Second * 10
-	}
-
-	var server = http.Server{
-		Handler:      h,
-		ReadTimeout:  h.ReadTimeout.Duration,
-		WriteTimeout: h.WriteTimeout.Duration,
-	}
-
-	return server.Serve(h.listener)
-}
-
 func (h *HTTPListener) ServeHTTP(res http.ResponseWriter, req *http.Request) {
 	h.RequestsRecv.Incr(1)
 	defer h.RequestsServed.Incr(1)
@@ -327,6 +349,48 @@ func badRequest(res http.ResponseWriter) {
 	res.Write([]byte(`{"error":"http: bad request"}`))
 }
 
+func (h *HTTPListener) getTLSConfig() *tls.Config {
+	tlsConf := &tls.Config{
+		InsecureSkipVerify: false,
+		Renegotiation:      tls.RenegotiateNever,
+	}
+
+	if len(h.SslCertificate) == 0 || len(h.SslKey) == 0 {
+		return nil
+	}
+
+	cert, err := tls.LoadX509KeyPair(h.SslCertificate, h.SslKey)
+	if err != nil {
+		return nil
+	}
+	tlsConf.Certificates = []tls.Certificate{cert}
+
+	roots := x509.NewCertPool()
+	for _, ca := range h.SslCertificateAuthorities {
+		c, err := ioutil.ReadFile(ca)
+		if err != nil {
+			continue
+		}
+		roots.AppendCertsFromPEM(c)
+	}
+	tlsConf.RootCAs = roots
+
+	if h.SslAllowedClientCertificateAuthorities != nil {
+		tlsConf.ClientAuth = tls.RequireAndVerifyClientCert
+		clientPool := x509.NewCertPool()
+		for _, ca := range h.SslAllowedClientCertificateAuthorities {
+			c, err := ioutil.ReadFile(ca)
+			if err != nil {
+				continue
+			}
+			clientPool.AppendCertsFromPEM(c)
+		}
+		tlsConf.ClientCAs = clientPool
+	}
+
+	return tlsConf
+}
+
 func init() {
 	inputs.Add("http_listener", func() telegraf.Input {
 		return &HTTPListener{