[xserver] TLS support added to xserver, aggregator server, and aggregator client

src/aggregator/client/config.go

@@ -207,6 +207,37 @@ func (c *Configuration) NewClientOptions(
 	return opts, nil
 }
 
+// TLSConfiguration contains the TLS configuration
+type TLSConfiguration struct {
+	TLSEnabled         bool    `yaml:"tlsEnabled"`
+	InsecureSkipVerify *bool   `yaml:"insecureSkipVerify"`
+	ServerName         *string `yaml:"serverName"`
+	CAFile             *string `yaml:"caFile"`
+	CertFile           *string `yaml:"certFile"`
+	KeyFile            *string `yaml:"keyFile"`
+}
+
+// NewTLSOptions creates new TLS options
+func (c *TLSConfiguration) NewTLSOptions() TLSOptions {
+	opts := NewTLSOptions().SetTLSEnabled(c.TLSEnabled)
+	if c.InsecureSkipVerify != nil {
+		opts = opts.SetInsecureSkipVerify(*c.InsecureSkipVerify)
+	}
+	if c.ServerName != nil {
+		opts = opts.SetServerName(*c.ServerName)
+	}
+	if c.CAFile != nil {
+		opts = opts.SetCAFile(*c.CAFile)
+	}
+	if c.CertFile != nil {
+		opts = opts.SetCertFile(*c.CertFile)
+	}
+	if c.KeyFile != nil {
+		opts = opts.SetKeyFile(*c.KeyFile)
+	}
+	return opts
+}
+
 // ConnectionConfiguration contains the connection configuration.
 type ConnectionConfiguration struct {
 	ConnectionTimeout            time.Duration        `yaml:"connectionTimeout"`
@@ -217,6 +248,7 @@ type ConnectionConfiguration struct {
 	ReconnectThresholdMultiplier int                  `yaml:"reconnectThresholdMultiplier"`
 	MaxReconnectDuration         *time.Duration       `yaml:"maxReconnectDuration"`
 	WriteRetries                 *retry.Configuration `yaml:"writeRetries"`
+	TLS                          *TLSConfiguration    `yaml:"tls"`
 }
 
 // NewConnectionOptions creates new connection options.
@@ -247,6 +279,9 @@ func (c *ConnectionConfiguration) NewConnectionOptions(scope tally.Scope) Connec
 		retryOpts := c.WriteRetries.NewOptions(scope)
 		opts = opts.SetWriteRetryOptions(retryOpts)
 	}
+	if c.TLS != nil {
+		opts = opts.SetTLSOptions(c.TLS.NewTLSOptions())
+	}
 	return opts
 }
 

src/aggregator/client/config_test.go

@@ -80,6 +80,13 @@ connection:
     maxBackoff: 1s
     maxRetries: 2
     jitter: true
+  tls:
+    tlsEnabled: false
+    insecureSkipVerify: true
+    serverName: TestServer
+    caFile: /tmp/ca
+    certFile: /tmp/cert
+    keyFile: /tmp/key
 `
 
 func TestConfigUnmarshal(t *testing.T) {
@@ -120,6 +127,12 @@ func TestConfigUnmarshal(t *testing.T) {
 	require.Equal(t, time.Second, cfg.Connection.WriteRetries.MaxBackoff)
 	require.Equal(t, 2, cfg.Connection.WriteRetries.MaxRetries)
 	require.Equal(t, true, *cfg.Connection.WriteRetries.Jitter)
+	require.False(t, cfg.Connection.TLS.TLSEnabled)
+	require.True(t, *cfg.Connection.TLS.InsecureSkipVerify)
+	require.Equal(t, "TestServer", *cfg.Connection.TLS.ServerName)
+	require.Equal(t, "/tmp/ca", *cfg.Connection.TLS.CAFile)
+	require.Equal(t, "/tmp/cert", *cfg.Connection.TLS.CertFile)
+	require.Equal(t, "/tmp/key", *cfg.Connection.TLS.KeyFile)
 	require.Nil(t, cfg.Connection.WriteRetries.Forever)
 }
 
@@ -171,4 +184,10 @@ func TestNewClientOptions(t *testing.T) {
 	require.Equal(t, 2, opts.ConnectionOptions().WriteRetryOptions().MaxRetries())
 	require.Equal(t, true, opts.ConnectionOptions().WriteRetryOptions().Jitter())
 	require.Equal(t, false, opts.ConnectionOptions().WriteRetryOptions().Forever())
+	require.False(t, opts.ConnectionOptions().TLSOptions().TLSEnabled())
+	require.True(t, opts.ConnectionOptions().TLSOptions().InsecureSkipVerify())
+	require.Equal(t, "TestServer", opts.ConnectionOptions().TLSOptions().ServerName())
+	require.Equal(t, "/tmp/ca", opts.ConnectionOptions().TLSOptions().CAFile())
+	require.Equal(t, "/tmp/cert", opts.ConnectionOptions().TLSOptions().CertFile())
+	require.Equal(t, "/tmp/key", opts.ConnectionOptions().TLSOptions().KeyFile())
 }

src/aggregator/client/conn.go

@@ -22,9 +22,13 @@ package client
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"errors"
+	"fmt"
 	"math/rand"
 	"net"
+	"os"
 	"sync"
 	"time"
 
@@ -77,6 +81,7 @@ type connection struct {
 	mtx                     sync.Mutex
 	keepAlive               bool
 	dialer                  xnet.ContextDialerFn
+	tls                     TLSOptions
 }
 
 // newConnection creates a new connection.
@@ -101,6 +106,7 @@ func newConnection(addr string, opts ConnectionOptions) *connection {
 			xio.ResettableWriterOptions{WriteBufferSize: 0},
 		),
 		metrics: newConnectionMetrics(opts.InstrumentOptions().MetricsScope()),
+		tls:     opts.TLSOptions(),
 	}
 	c.connectWithLockFn = c.connectWithLock
 	c.writeWithLockFn = c.writeWithLock
@@ -152,6 +158,34 @@ func (c *connection) Close() {
 	c.mtx.Unlock()
 }
 
+func (c *connection) upgradeToTLS(conn net.Conn) (net.Conn, error) {
+	certPool := x509.NewCertPool()
+	if c.tls.CAFile() != "" {
+		certs, err := os.ReadFile(c.tls.CAFile())
+		if err != nil {
+			return conn, fmt.Errorf("read bundle error: %w", err)
+		}
+		if ok := certPool.AppendCertsFromPEM(certs); !ok {
+			return conn, fmt.Errorf("cannot append cert to cert pool")
+		}
+	}
+	tlsConfig := &tls.Config{
+		RootCAs:            certPool,
+		InsecureSkipVerify: c.tls.InsecureSkipVerify(),
+		ServerName:         c.tls.ServerName(),
+	}
+	if c.tls.CertFile() != "" && c.tls.KeyFile() != "" {
+		tlsConfig.GetClientCertificate = func(*tls.CertificateRequestInfo) (*tls.Certificate, error) {
+			cert, err := tls.LoadX509KeyPair(c.tls.CertFile(), c.tls.KeyFile())
+			if err != nil {
+				return nil, fmt.Errorf("load x509 key pair error: %w", err)
+			}
+			return &cert, nil
+		}
+	}
+	return tls.Client(conn, tlsConfig), nil
+}
+
 // writeAttemptWithLock attempts to establish a new connection and writes raw bytes
 // to the connection while holding the write lock.
 // If the write succeeds, c.conn is guaranteed to be a valid connection on return.
@@ -192,6 +226,15 @@ func (c *connection) connectWithLock() error {
 		}
 	}
 
+	if c.tls.TLSEnabled() {
+		conn, err = c.upgradeToTLS(conn)
+		if err != nil {
+			c.metrics.connectError.Inc(1)
+			conn.Close()
+			return err
+		}
+	}
+
 	if c.conn != nil {
 		c.conn.Close() // nolint: errcheck
 	}

src/aggregator/client/conn_options.go

@@ -113,6 +113,12 @@ type ConnectionOptions interface {
 	// RWOptions returns the RW options.
 	RWOptions() xio.Options
 
+	// SetTLSOptions sets TLS options
+	SetTLSOptions(value TLSOptions) ConnectionOptions
+
+	// TLSOptions returns the TLS options
+	TLSOptions() TLSOptions
+
 	// ContextDialer allows customizing the way an aggregator client the aggregator, at the TCP layer.
 	// By default, this is:
 	// (&net.ContextDialer{}).DialContext. This can be used to do a variety of things, such as forwarding a connection
@@ -137,6 +143,7 @@ type connectionOptions struct {
 	maxThreshold   int
 	multiplier     int
 	connKeepAlive  bool
+	tlsOptions     TLSOptions
 	dialer         xnet.ContextDialerFn
 }
 
@@ -159,6 +166,7 @@ func NewConnectionOptions() ConnectionOptions {
 		multiplier:     defaultReconnectThresholdMultiplier,
 		maxDuration:    defaultMaxReconnectDuration,
 		writeRetryOpts: defaultWriteRetryOpts,
+		tlsOptions:     NewTLSOptions(),
 		rwOpts:         xio.NewOptions(),
 		dialer:         nil, // Will default to net.Dialer{}.DialContext
 	}
@@ -274,6 +282,16 @@ func (o *connectionOptions) RWOptions() xio.Options {
 	return o.rwOpts
 }
 
+func (o *connectionOptions) SetTLSOptions(value TLSOptions) ConnectionOptions {
+	opts := *o
+	opts.tlsOptions = value
+	return &opts
+}
+
+func (o *connectionOptions) TLSOptions() TLSOptions {
+	return o.tlsOptions
+}
+
 func (o *connectionOptions) ContextDialer() xnet.ContextDialerFn {
 	return o.dialer
 }

src/aggregator/client/conn_test.go

@@ -22,10 +22,13 @@ package client
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"errors"
 	"fmt"
 	"math"
 	"net"
+	"os"
 	"sync"
 	"testing"
 	"time"
@@ -405,6 +408,76 @@ func TestConnectWriteToServer(t *testing.T) {
 	require.Nil(t, conn.conn)
 }
 
+func TestTLSConnectWriteToServer(t *testing.T) {
+	data := []byte("foobar")
+
+	// Start tls server.
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	serverCert, err := tls.LoadX509KeyPair("./testdata/server.crt", "./testdata/server.key")
+	require.NoError(t, err)
+	certPool := x509.NewCertPool()
+	certs, err := os.ReadFile("./testdata/rootCA.crt")
+	require.NoError(t, err)
+	certPool.AppendCertsFromPEM(certs)
+	l, err := tls.Listen(tcpProtocol, testLocalServerAddr, &tls.Config{
+		Certificates: []tls.Certificate{serverCert},
+		ClientCAs:    certPool,
+		ClientAuth:   tls.RequireAndVerifyClientCert,
+	})
+	require.NoError(t, err)
+	serverAddr := l.Addr().String()
+
+	go func() {
+		defer wg.Done()
+
+		// Ignore the first testing connection.
+		conn, err := l.Accept()
+		tlsConn, ok := conn.(*tls.Conn)
+		require.True(t, ok)
+		tlsConn.Handshake()
+		require.NoError(t, err)
+		require.NoError(t, conn.Close())
+
+		// Read from the second connection.
+		conn, err = l.Accept()
+		require.NoError(t, err)
+		buf := make([]byte, 1024)
+		n, err := conn.Read(buf)
+		require.NoError(t, err)
+		require.Equal(t, data, buf[:n])
+		conn.Close() // nolint: errcheck
+	}()
+
+	clientCert, err := tls.LoadX509KeyPair("./testdata/client.crt", "./testdata/client.key")
+	require.NoError(t, err)
+	// Wait until the server starts up.
+	dialer := net.Dialer{Timeout: time.Minute}
+	testConn, err := tls.DialWithDialer(&dialer, tcpProtocol, serverAddr, &tls.Config{
+		InsecureSkipVerify: true,
+		Certificates:       []tls.Certificate{clientCert},
+		RootCAs:            certPool,
+	})
+	require.NoError(t, err)
+	require.NoError(t, testConn.Close())
+
+	// Create a new connection and assert we can write successfully.
+	opts := testTLSConnectionOptions().SetInitReconnectThreshold(0)
+	conn := newConnection(serverAddr, opts)
+	require.NoError(t, conn.Write(data))
+	require.Equal(t, 0, conn.numFailures)
+	require.NotNil(t, conn.conn)
+
+	// Stop the server.
+	l.Close() // nolint: errcheck
+	wg.Wait()
+
+	// Close the connection
+	conn.Close()
+	require.Nil(t, conn.conn)
+}
+
 func testConnectionOptions() ConnectionOptions {
 	return NewConnectionOptions().
 		SetClockOptions(clock.NewOptions()).
@@ -416,6 +489,16 @@ func testConnectionOptions() ConnectionOptions {
 		SetWriteTimeout(100 * time.Millisecond)
 }
 
+func testTLSConnectionOptions() ConnectionOptions {
+	tlsOptions := NewTLSOptions().
+		SetTLSEnabled(true).
+		SetInsecureSkipVerify(true).
+		SetCAFile("./testdata/rootCA.crt").
+		SetCertFile("./testdata/client.crt").
+		SetKeyFile("./testdata/client.key")
+	return testConnectionOptions().SetTLSOptions(tlsOptions)
+}
+
 func testConnectionProperties() *gopter.Properties {
 	params := gopter.DefaultTestParameters()
 	params.Rng.Seed(testRandomSeeed)

src/aggregator/client/options.go

@@ -251,6 +251,7 @@ type options struct {
 	maxBatchSize               int
 	flushWorkerCount           int
 	aggregatorClientType       AggregatorClientType
+	tlsOptions                 TLSOptions
 }
 
 // NewOptions creates a new set of client options.

src/aggregator/client/testdata/client.crt

@@ -0,0 +1,20 @@
+-----BEGIN CERTIFICATE-----
+MIIDPzCCAicCFFJVDW2T6lamcRZYStsBCQTKicKwMA0GCSqGSIb3DQEBCwUAMFox
+CzAJBgNVBAYTAk5MMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRl
+cm5ldCBXaWRnaXRzIFB0eSBMdGQxEzARBgNVBAMMClRlc3RSb290Q0EwHhcNMjQw
+NDE1MTQ0NjEyWhcNMzQwNDEzMTQ0NjEyWjBeMQswCQYDVQQGEwJOTDETMBEGA1UE
+CAwKU29tZS1TdGF0ZTEhMB8GA1UECgwYSW50ZXJuZXQgV2lkZ2l0cyBQdHkgTHRk
+MRcwFQYDVQQDDA5UZXN0Q2xpZW50Q2VydDCCASIwDQYJKoZIhvcNAQEBBQADggEP
+ADCCAQoCggEBAMXgv5tgZD5iBW3LfTC27jPHTLlsAw8ZlBFLmd7potdS3Hy512Xa
+kKOaKRKH/0Yla4BKGNh0ZP+xege15CpTFqwtzDPCRkaRvfSxRW19MukeergRDqqI
+tEwURHQ5gfub7XB9e5wuNmW2GGs/xgiWsxxm6UMidXkhA5a2//OmDSm2JdcoxJbk
+1m1OUzXUuLKLzCIRAbJFFIM1RpvjWSonxpJ2lsNCVQ+LdItxE9mAI+Y+PaDonyzg
+ehfAqkl+wxGRevaDFFJr1defxmnJhaYv40AcEpXOWeetZFlspEtrPF7b/3GifRdx
+Mlwq27wn3mk4wF1wa12kRNKiz+HHjZrB7UECAwEAATANBgkqhkiG9w0BAQsFAAOC
+AQEAKblZ+C4Pdv5grKiCQcfFt53SbIys8uavxmzJNn9nq1QVgBmR/hX8yG8ULeh5
+kKrccwe2WXCBWsIj9cvGCzR+MPMvuIXd3d58yLybmIUBDfJSBC0v5TaNWP2ZdJDG
+fH38DfQTeE9XcjpRtCGYtYwReBptSlNiYQlvwkukTNz3mV+0BpvQu/9/R0BNCJAF
+9VeLgUcG7T7HC+foCeU7h83Y0xH0OALe33ntcrMPHUclXqImCiC7dl8pb3H7o/xJ
+jS3KZxfv6ioxpirvpEbX/EeF+H1HQZUZ8y3+J3K37jMgzNamd/xsaJxEfCq4SE+n
+HmgeXgqoNjrM6J0jYAfx1MjbIA==
+-----END CERTIFICATE-----

src/aggregator/client/testdata/client.key

@@ -0,0 +1,27 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIEowIBAAKCAQEAxeC/m2BkPmIFbct9MLbuM8dMuWwDDxmUEUuZ3umi11LcfLnX
+ZdqQo5opEof/RiVrgEoY2HRk/7F6B7XkKlMWrC3MM8JGRpG99LFFbX0y6R56uBEO
+qoi0TBREdDmB+5vtcH17nC42ZbYYaz/GCJazHGbpQyJ1eSEDlrb/86YNKbYl1yjE
+luTWbU5TNdS4sovMIhEBskUUgzVGm+NZKifGknaWw0JVD4t0i3ET2YAj5j49oOif
+LOB6F8CqSX7DEZF69oMUUmvV15/GacmFpi/jQBwSlc5Z561kWWykS2s8Xtv/caJ9
+F3EyXCrbvCfeaTjAXXBrXaRE0qLP4ceNmsHtQQIDAQABAoIBAAtyP8MuJT5SjzvV
+rI0317mZCsAjFl42PZFujR0O6MOJ4IU6ftI+fWVpUnzm7wZQvdIy9xL2UK1Vx9hQ
+Vj14hvQ4xfosf8IvRgy0gG6f8mT3xWOGYRHOTJemCHuso+85CtgZ+h+DsNPbX7g8
+fSkcBopbDZ07jg4OsdVzCoU+kr5Z1/VDe0O0rTDzGVP686tN9I+DmDYv2HlBUaSZ
+SyNZrLJFziqFx4ATTIE3aTLd29pzTl09WiCcZSxhlS+ROnN2iW2xDLNgY9SdUo2a
+3r8N+xhYDNQz2paxlsv2x/tCRJvrQDoX7S5QZw4P1pVr6wo6xC+BU2UlpVHifo4M
+egyDbAECgYEA5Y9+JDOSQF2cvIV4xTMgVnkLig/ngYesS44xz8HYTRR5s1WW6oCX
+3j+OosZBbJpXeBlTK31G9Z92+Om2/Cj2uDvff7EJdhZ5sAdMtUy5X7p9rGTEyi5F
+ecZDBfmwHH0lFdwpNtlky1uoaUsVd4IS28WMHoHYDGW241IW8xOlmeECgYEA3Ksb
+KR7JCbHx21wh5oPVn5BLpdag7D+M1jBFtjPrzLDNhXlsV9zkjn7xETvTXWU2vNlh
+OkoRu4VFEAnsgtxh+gT0lo9ZG9IWi2qP99vPpB3AFYuujBu3GA02/iOo5pri/6R0
+ZNEEreAaaOcT2z6K9HG0KMQ6QpYgXSlzJW8sf2ECgYEAvn2fMA03bIAB4xJi0EkH
+qZoScEOYWQ0rdRsOzJbPlc7K2nzImdmRrFRTWVFo0uUUdk2VjX4Mlx/3ir/uHzsi
+2GieowhWkI4/9kloZv2+yegoBxkrj5ZsAov57AhxEoLqdkRWUvR8xp9Nleo/awcd
+/Q7loh8fF9KDvAjPkHAaOCECgYA98bZNI7wxgYcwGbvWdrmX8iyaIBbKWsiRM7nN
+/OM7cYIv7rbwLyzlp1LKkK2zsP7donP9pd82caHCb9a5oV3LjmqOfSz5d08m0cIa
+RNUT79oE8lIMOJd8I/GFA8OdAGuqcaLOzjHvEVK4ke1sBTGCjwyQyQzFtljdbg5J
+utyV4QKBgHkSBKwrgaump08MJFfK9uFo5sr5jy692XYWMlBVhc+piPwUsoztjI1O
+8cvie0mMj6oofSoSyD4FUg9kFUCRMPC0kga/zlHyL1Y/uILX4XUFMl4P/gMbC7kw
+0d6BdEsp5SU8Onc9eV/EW+q2Mz7/t5ZHkUrdDQYGgkIBqRZcXlka
+-----END RSA PRIVATE KEY-----