Add TLS support to socket_writer and socket_listener plugins

internal/internal.go

@@ -112,12 +112,21 @@ func RandomString(n int) string {
 	return string(bytes)
 }
 
-// GetTLSConfig gets a tls.Config object from the given certs, key, and CA files.
-// you must give the full path to the files.
-// If all files are blank and InsecureSkipVerify=false, returns a nil pointer.
+// Deprecated - use GetClientTLSConfig or GetServerTLSConfig instead.
 func GetTLSConfig(
 	SSLCert, SSLKey, SSLCA string,
 	InsecureSkipVerify bool,
+) (*tls.Config, error) {
+	return GetClientTLSConfig(SSLCert, SSLKey, SSLCA, InsecureSkipVerify)
+}
+
+// GetClientTLSConfig gets a tls.Config object from the given certs, key, and CA files
+// for use with a client.
+// The full path to each file must be provided.
+// Returns a nil pointer if all files are blank and InsecureSkipVerify=false.
+func GetClientTLSConfig(
+	SSLCert, SSLKey, SSLCA string,
+	InsecureSkipVerify bool,
 ) (*tls.Config, error) {
 	if SSLCert == "" && SSLKey == "" && SSLCA == "" && !InsecureSkipVerify {
 		return nil, nil
@@ -155,6 +164,53 @@ func GetTLSConfig(
 	return t, nil
 }
 
+// GetServerTLSConfig gets a tls.Config object from the given certs, key, and one or more CA files
+// for use with a server.
+// The full path to each file must be provided.
+// Returns a nil pointer if all files are blank.
+func GetServerTLSConfig(
+	SSLCert, SSLKey string,
+	SSLCA []string,
+	SSLClientAuth bool,
+) (*tls.Config, error) {
+	if SSLCert == "" && SSLKey == "" && len(SSLCA) == 0 {
+		return nil, nil
+	}
+
+	t := &tls.Config{}
+
+	if len(SSLCA) != 0 {
+		caCertPool := x509.NewCertPool()
+		for _, cert := range SSLCA {
+			c, err := ioutil.ReadFile(cert)
+			if err != nil {
+				return nil, errors.New(fmt.Sprintf("Could not load TLS CA: %s",
+					err))
+			}
+			caCertPool.AppendCertsFromPEM(c)
+		}
+		t.ClientCAs = caCertPool
+	}
+
+	if SSLCert != "" && SSLKey != "" {
+		cert, err := tls.LoadX509KeyPair(SSLCert, SSLKey)
+		if err != nil {
+			return nil, errors.New(fmt.Sprintf(
+				"Could not load TLS client key/certificate from %s:%s: %s",
+				SSLKey, SSLCert, err))
+		}
+
+		t.Certificates = []tls.Certificate{cert}
+		t.BuildNameToCertificate()
+	}
+
+	if SSLClientAuth {
+		t.ClientAuth = tls.RequireAndVerifyClientCert
+	}
+
+	return t, nil
+}
+
 // SnakeCase converts the given string to snake case following the Golang format:
 // acronyms are converted to lower-case and preceded by an underscore.
 func SnakeCase(in string) string {

plugins/inputs/socket_listener/README.md

@@ -35,6 +35,15 @@ This is a sample configuration for the plugin.
   ## 0 (default) is unlimited.
   # read_timeout = "30s"
 
+  ## Optional SSL configuration.
+  ## Only applies to stream sockets (e.g. TCP).
+  # ssl_cert = "/etc/telegraf/cert.pem"
+  # ssl_key = "/etc/telegraf/key.pem"
+  ## Enable and require client certificate authentication.
+  # ssl_client_auth = false
+  ## CAs used to verify client certificates.
+  # ssl_ca = ["/etc/telegraf/ca.pem"]
+
   ## Maximum socket buffer size in bytes.
   ## For stream sockets, once the buffer fills up, the sender will start backing up.
   ## For datagram sockets, once the buffer fills up, metrics will start dropping.

plugins/inputs/socket_listener/socket_listener.go

@@ -12,6 +12,7 @@ import (
 
 	"time"
 
+	"crypto/tls"
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/plugins/inputs"
@@ -163,6 +164,10 @@ type SocketListener struct {
 	MaxConnections  int
 	ReadBufferSize  int
 	ReadTimeout     *internal.Duration
+	SSLCA           []string
+	SSLCert         string
+	SSLKey          string
+	SSLClientAuth   bool
 	KeepAlivePeriod *internal.Duration
 
 	parsers.Parser
@@ -198,6 +203,15 @@ func (sl *SocketListener) SampleConfig() string {
   ## 0 (default) is unlimited.
   # read_timeout = "30s"
 
+  ## Optional SSL configuration.
+  ## Only applies to stream sockets (e.g. TCP).
+  # ssl_cert = "/etc/telegraf/cert.pem"
+  # ssl_key = "/etc/telegraf/key.pem"
+  ## Enable and require client certificate authentication.
+  # ssl_client_auth = false
+  ## CAs used to verify client certificates.
+  # ssl_ca = ["/etc/telegraf/ca.pem"]
+
   ## Maximum socket buffer size in bytes.
   ## For stream sockets, once the buffer fills up, the sender will start backing up.
   ## For datagram sockets, once the buffer fills up, metrics will start dropping.
@@ -242,7 +256,21 @@ func (sl *SocketListener) Start(acc telegraf.Accumulator) error {
 
 	switch spl[0] {
 	case "tcp", "tcp4", "tcp6", "unix", "unixpacket":
-		l, err := net.Listen(spl[0], spl[1])
+		var (
+			err error
+			l   net.Listener
+		)
+
+		tlsCfg, err := internal.GetServerTLSConfig(sl.SSLCert, sl.SSLKey, sl.SSLCA, sl.SSLClientAuth)
+		if err != nil {
+			return nil
+		}
+
+		if tlsCfg == nil {
+			l, err = net.Listen(spl[0], spl[1])
+		} else {
+			l, err = tls.Listen(spl[0], spl[1], tlsCfg)
+		}
 		if err != nil {
 			return err
 		}

plugins/outputs/socket_writer/README.md

@@ -19,6 +19,13 @@ It can output data in any of the [supported output formats](https://github.com/i
   # address = "unix:///tmp/telegraf.sock"
   # address = "unixgram:///tmp/telegraf.sock"
 
+  ## Optional SSL Config
+  # ssl_ca = "/etc/telegraf/ca.pem"
+  # ssl_cert = "/etc/telegraf/cert.pem"
+  # ssl_key = "/etc/telegraf/key.pem"
+  ## Use SSL but skip chain & host verification
+  # insecure_skip_verify = false
+
   ## Period between keep alive probes.
   ## Only applies to TCP sockets.
   ## 0 disables keep alive probes.

plugins/outputs/socket_writer/socket_writer.go

@@ -6,15 +6,20 @@ import (
 	"net"
 	"strings"
 
+	"crypto/tls"
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/plugins/outputs"
 	"github.com/influxdata/telegraf/plugins/serializers"
 )
 
 type SocketWriter struct {
-	Address         string
-	KeepAlivePeriod *internal.Duration
+	Address            string
+	KeepAlivePeriod    *internal.Duration
+	SSLCA              string
+	SSLCert            string
+	SSLKey             string
+	InsecureSkipVerify bool
 
 	serializers.Serializer
 
@@ -39,6 +44,13 @@ func (sw *SocketWriter) SampleConfig() string {
   # address = "unix:///tmp/telegraf.sock"
   # address = "unixgram:///tmp/telegraf.sock"
 
+  ## Optional SSL Config
+  # ssl_ca = "/etc/telegraf/ca.pem"
+  # ssl_cert = "/etc/telegraf/cert.pem"
+  # ssl_key = "/etc/telegraf/key.pem"
+  ## Use SSL but skip chain & host verification
+  # insecure_skip_verify = false
+
   ## Period between keep alive probes.
   ## Only applies to TCP sockets.
   ## 0 disables keep alive probes.
@@ -58,12 +70,26 @@ func (sw *SocketWriter) SetSerializer(s serializers.Serializer) {
 }
 
 func (sw *SocketWriter) Connect() error {
+	var (
+		c   net.Conn
+		err error
+	)
+
 	spl := strings.SplitN(sw.Address, "://", 2)
 	if len(spl) != 2 {
 		return fmt.Errorf("invalid address: %s", sw.Address)
 	}
 
-	c, err := net.Dial(spl[0], spl[1])
+	tlsCfg, err := internal.GetClientTLSConfig(sw.SSLCert, sw.SSLKey, sw.SSLCA, sw.InsecureSkipVerify)
+	if err != nil {
+		return err
+	}
+
+	if tlsCfg == nil {
+		c, err = net.Dial(spl[0], spl[1])
+	} else {
+		c, err = tls.Dial(spl[0], spl[1], tlsCfg)
+	}
 	if err != nil {
 		return err
 	}