Add client TLS auth to gRPC reporter

cmd/agent/app/reporter/grpc/builder.go

@@ -15,7 +15,11 @@
 package grpc
 
 import (
+	"crypto/tls"
+	"crypto/x509"
 	"errors"
+	"fmt"
+	"io/ioutil"
 	"strings"
 
 	grpc_retry "github.com/grpc-ecosystem/go-grpc-middleware/retry"
@@ -39,6 +43,8 @@ type ConnBuilder struct {
 	TLS           bool
 	TLSCA         string
 	TLSServerName string
+	TLSCert       string
+	TLSKey        string
 
 	DiscoveryMinPeers int
 	Notifier          discovery.Notifier
@@ -56,20 +62,47 @@ func (b *ConnBuilder) CreateConnection(logger *zap.Logger) (*grpc.ClientConn, er
 	var dialTarget string
 	if b.TLS { // user requested a secure connection
 		logger.Info("Agent requested secure grpc connection to collector(s)")
-		var creds credentials.TransportCredentials
+		var err error
+		var certPool *x509.CertPool
 		if len(b.TLSCA) == 0 { // no truststore given, use SystemCertPool
-			pool, err := systemCertPool()
+			certPool, err = systemCertPool()
 			if err != nil {
 				return nil, err
 			}
-			creds = credentials.NewClientTLSFromCert(pool, b.TLSServerName)
 		} else { // setup user specified truststore
-			var err error
-			creds, err = credentials.NewClientTLSFromFile(b.TLSCA, b.TLSServerName)
+			caPEM, err := ioutil.ReadFile(b.TLSCA)
 			if err != nil {
-				return nil, err
+				return nil, fmt.Errorf("reading client CA failed, %v", err)
+			}
+
+			certPool = x509.NewCertPool()
+			if !certPool.AppendCertsFromPEM(caPEM) {
+				return nil, fmt.Errorf("building client CA failed, %v", err)
 			}
 		}
+
+		tlsCfg := &tls.Config{
+			MinVersion: tls.VersionTLS12,
+			RootCAs:    certPool,
+			ServerName: b.TLSServerName,
+		}
+
+		if (b.TLSKey == "" || b.TLSCert == "") &&
+			(b.TLSKey != "" || b.TLSCert != "") {
+			return nil, fmt.Errorf("for client auth, both client certificate and key must be supplied")
+		}
+
+		if b.TLSKey != "" && b.TLSCert != "" {
+			tlsCert, err := tls.LoadX509KeyPair(b.TLSCert, b.TLSKey)
+			if err != nil {
+				return nil, fmt.Errorf("could not load server TLS cert and key, %v", err)
+			}
+
+			logger.Info("client TLS authentication enabled")
+			tlsCfg.Certificates = []tls.Certificate{tlsCert}
+		}
+
+		creds := credentials.NewTLS(tlsCfg)
 		dialOptions = append(dialOptions, grpc.WithTransportCredentials(creds))
 	} else { // insecure connection
 		logger.Info("Agent requested insecure grpc connection to collector(s)")

cmd/agent/app/reporter/grpc/builder_test.go

@@ -15,35 +15,26 @@ package grpc
 
 import (
 	"errors"
-	"io/ioutil"
-	"os"
+	"net"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/uber/jaeger-lib/metrics"
+	"github.com/uber/jaeger-lib/metrics/metricstest"
 	"go.uber.org/zap"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
 	yaml "gopkg.in/yaml.v2"
 
+	"github.com/jaegertracing/jaeger/cmd/collector/app/grpcserver"
 	"github.com/jaegertracing/jaeger/pkg/discovery"
+	"github.com/jaegertracing/jaeger/proto-gen/api_v2"
+	"github.com/jaegertracing/jaeger/thrift-gen/jaeger"
 )
 
-const certPEM = `
------BEGIN CERTIFICATE-----
-MIICBzCCAXCgAwIBAgIQNkTaUtOczDHvL2YT/kqScTANBgkqhkiG9w0BAQsFADAX
-MRUwEwYDVQQKEwxqYWdlcnRyYWNpbmcwHhcNMTkwMjA4MDYyODAyWhcNMTkwMjA4
-MDcyODAyWjAXMRUwEwYDVQQKEwxqYWdlcnRyYWNpbmcwgZ8wDQYJKoZIhvcNAQEB
-BQADgY0AMIGJAoGBAMcOLYflHGbqC1f7+tbnsdfcpd0rEuX65+ab0WzelAgvo988
-yD+j7LDLPIE8IPk/tfqaETZ8h0LRUUTn8F2rW/wgrl/G8Onz0utog38N0elfTifG
-Mu7GJCr/+aYM5xbQMDj4Brb4vhnkJF8UBe49fWILhIltUcm1SeKqVX3d1FvpAgMB
-AAGjVDBSMA4GA1UdDwEB/wQEAwICpDATBgNVHSUEDDAKBggrBgEFBQcDATAPBgNV
-HRMBAf8EBTADAQH/MBoGA1UdEQQTMBGCCWxvY2FsaG9zdIcEfwAAATANBgkqhkiG
-9w0BAQsFAAOBgQCreFjwpAn1HqJT812JOwoWKrt1NjOKGcz7pvIs1k3DfQVLH2aZ
-iPKnCkzNgxMzQtwdgpAOXIAqXyNibvyOAv1C+3QSMLKbuPEHaIxlCuvl1suX/g25
-17x1o3Q64AnPCWOLpN2wjkfZqX7gZ84nsxpqb9Sbw1+2+kqX7dSZ3mfVxQ==
------END CERTIFICATE-----`
-
 var yamlConfig = `
 collectorHostPorts:
     - 127.0.0.1:14267
@@ -141,20 +132,6 @@ func TestBuilderWithCollectors(t *testing.T) {
 }
 
 func TestProxyBuilder(t *testing.T) {
-	tmpfile, err := ioutil.TempFile("", "cert*.pem")
-	if err != nil {
-		t.Fatalf("failed to create tempfile: %s", err)
-	}
-
-	defer func() {
-		tmpfile.Close()
-		os.Remove(tmpfile.Name())
-	}()
-
-	if _, err := tmpfile.Write([]byte(certPEM)); err != nil {
-		t.Fatalf("failed to write cert to tempfile: %s", err)
-	}
-
 	tests := []struct {
 		name        string
 		grpcBuilder *ConnBuilder
@@ -172,14 +149,47 @@ func TestProxyBuilder(t *testing.T) {
 		},
 		{
 			name:        "with secure grpc connection and own CA",
-			grpcBuilder: &ConnBuilder{CollectorHostPorts: []string{"localhost:0000"}, TLS: true, TLSCA: tmpfile.Name()},
+			grpcBuilder: &ConnBuilder{CollectorHostPorts: []string{"localhost:0000"}, TLS: true, TLSCA: "testdata/testCA.pem"},
 			expectError: false,
 		},
 		{
 			name:        "with secure grpc connection and a CA file which does not exist",
 			grpcBuilder: &ConnBuilder{CollectorHostPorts: []string{"localhost:0000"}, TLS: true, TLSCA: "/not/valid"},
 			expectError: true,
 		},
+		{
+			name: "with secure grpc connection and valid TLS Client settings",
+			grpcBuilder: &ConnBuilder{
+				CollectorHostPorts: []string{"localhost:0000"},
+				TLS:                true,
+				TLSCA:              "testdata/testCA.pem",
+				TLSCert:            "testdata/client.jaeger.io-client.pem",
+				TLSKey:             "testdata/client.jaeger.io-client-key.pem",
+			},
+			expectError: false,
+		},
+		{
+			name: "with secure grpc connection and valid TLS Client settings",
+			grpcBuilder: &ConnBuilder{
+				CollectorHostPorts: []string{"localhost:0000"},
+				TLS:                true,
+				TLSCA:              "testdata/testCA.pem",
+				TLSCert:            "testdata/client.jaeger.io-client.pem",
+				TLSKey:             "",
+			},
+			expectError: true,
+		},
+		{
+			name: "with secure grpc connection and valid TLS Client key setting",
+			grpcBuilder: &ConnBuilder{
+				CollectorHostPorts: []string{"localhost:0000"},
+				TLS:                true,
+				TLSCA:              "testdata/testCA.pem",
+				TLSCert:            "testdata/client.jaeger.io-client.pem",
+				TLSKey:             "/not/valid",
+			},
+			expectError: true,
+		},
 	}
 
 	for _, test := range tests {
@@ -200,3 +210,156 @@ func TestProxyBuilder(t *testing.T) {
 		})
 	}
 }
+
+func TestProxyClientTLS(t *testing.T) {
+	tests := []struct {
+		name              string
+		grpcBuilder       *ConnBuilder
+		serverTLS         bool
+		serverTLSCert     string
+		serverTLSKey      string
+		serverTLSClientCA string
+		expectError       bool
+	}{
+		{
+			name:        "insecure grpc connection",
+			serverTLS:   false,
+			grpcBuilder: &ConnBuilder{},
+			expectError: false,
+		},
+		{
+			name: "TLS client to non-TLS server should fail",
+			grpcBuilder: &ConnBuilder{
+				TLS: true,
+			},
+			expectError: true,
+		},
+		{
+			name:          "TLS client to untrusted TLS server should fail",
+			serverTLS:     true,
+			serverTLSCert: "testdata/server.jaeger.io.pem",
+			serverTLSKey:  "testdata/server.jaeger.io-key.pem",
+			grpcBuilder: &ConnBuilder{
+				TLS:           true,
+				TLSServerName: "server.jaeger.io",
+			},
+			expectError: true,
+		},
+		{
+			name:          "TLS client to trusted TLS server with incorrect hostname should fail",
+			serverTLS:     true,
+			serverTLSCert: "testdata/server.jaeger.io.pem",
+			serverTLSKey:  "testdata/server.jaeger.io-key.pem",
+			grpcBuilder: &ConnBuilder{
+				TLS:   true,
+				TLSCA: "testdata/rootCA.pem",
+			},
+			expectError: true,
+		},
+		{
+			name:          "TLS client to trusted TLS server with correct hostname",
+			serverTLS:     true,
+			serverTLSCert: "testdata/server.jaeger.io.pem",
+			serverTLSKey:  "testdata/server.jaeger.io-key.pem",
+			grpcBuilder: &ConnBuilder{
+				TLS:           true,
+				TLSCA:         "testdata/rootCA.pem",
+				TLSServerName: "server.jaeger.io",
+			},
+			expectError: false,
+		},
+		{
+			name:              "TLS client without cert to trusted TLS server requiring cert should fail",
+			serverTLS:         true,
+			serverTLSCert:     "testdata/server.jaeger.io.pem",
+			serverTLSKey:      "testdata/server.jaeger.io-key.pem",
+			serverTLSClientCA: "testdata/rootCA.pem",
+			grpcBuilder: &ConnBuilder{
+				TLS:           true,
+				TLSCA:         "testdata/rootCA.pem",
+				TLSServerName: "server.jaeger.io",
+			},
+			expectError: true,
+		},
+		{
+			name:              "TLS client without cert to trusted TLS server requiring cert from a differe CA should fail",
+			serverTLS:         true,
+			serverTLSCert:     "testdata/server.jaeger.io.pem",
+			serverTLSKey:      "testdata/server.jaeger.io-key.pem",
+			serverTLSClientCA: "testdata/testCA.pem",
+			grpcBuilder: &ConnBuilder{
+				TLS:           true,
+				TLSCA:         "testdata/rootCA.pem",
+				TLSServerName: "server.jaeger.io",
+				TLSCert:       "testdata/client.jaeger.io-client.pem",
+				TLSKey:        "testdata/client.jaeger.io-client-key.pem",
+			},
+			expectError: true,
+		},
+		{
+			name:              "TLS client without cert to trusted TLS server requiring cert should fail",
+			serverTLS:         true,
+			serverTLSCert:     "testdata/server.jaeger.io.pem",
+			serverTLSKey:      "testdata/server.jaeger.io-key.pem",
+			serverTLSClientCA: "testdata/rootCA.pem",
+			grpcBuilder: &ConnBuilder{
+				TLS:           true,
+				TLSCA:         "testdata/rootCA.pem",
+				TLSServerName: "server.jaeger.io",
+				TLSCert:       "testdata/client.jaeger.io-client.pem",
+				TLSKey:        "testdata/client.jaeger.io-client-key.pem",
+			},
+			expectError: false,
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			var opts []grpc.ServerOption
+			if test.serverTLS {
+				tlsCfg, err := grpcserver.TLSConfig(
+					test.serverTLSCert,
+					test.serverTLSKey,
+					test.serverTLSClientCA)
+				if err != nil {
+					require.NoError(t, err)
+				}
+
+				opts = []grpc.ServerOption{grpc.Creds(credentials.NewTLS(tlsCfg))}
+			}
+
+			spanHandler := &mockSpanHandler{}
+			s, addr := initializeGRPCTestServer(t, func(s *grpc.Server) {
+				api_v2.RegisterCollectorServiceServer(s, spanHandler)
+			}, opts...)
+			defer s.Stop()
+
+			mFactory := metricstest.NewFactory(time.Microsecond)
+			_, port, _ := net.SplitHostPort(addr.String())
+
+			test.grpcBuilder.CollectorHostPorts = []string{net.JoinHostPort("localhost", port)}
+			proxy, err := NewCollectorProxy(
+				test.grpcBuilder,
+				nil,
+				mFactory,
+				zap.NewNop())
+
+			require.NoError(t, err)
+			require.NotNil(t, proxy)
+			assert.NotNil(t, proxy.GetReporter())
+			assert.NotNil(t, proxy.GetManager())
+			assert.NotNil(t, proxy.GetConn())
+
+			r := proxy.GetReporter()
+
+			err = r.EmitBatch(&jaeger.Batch{Spans: []*jaeger.Span{{OperationName: "op"}}, Process: &jaeger.Process{ServiceName: "service"}})
+
+			if test.expectError {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+
+			require.Nil(t, proxy.Close())
+		})
+	}
+}

cmd/agent/app/reporter/grpc/collector_proxy_test.go

@@ -82,8 +82,8 @@ func TestMultipleCollectors(t *testing.T) {
 	require.Nil(t, proxy.Close())
 }
 
-func initializeGRPCTestServer(t *testing.T, beforeServe func(server *grpc.Server)) (*grpc.Server, net.Addr) {
-	server := grpc.NewServer()
+func initializeGRPCTestServer(t *testing.T, beforeServe func(server *grpc.Server), opts ...grpc.ServerOption) (*grpc.Server, net.Addr) {
+	server := grpc.NewServer(opts...)
 	lis, err := net.Listen("tcp", "localhost:0")
 	require.NoError(t, err)
 	beforeServe(server)

cmd/agent/app/reporter/grpc/flags.go

@@ -28,17 +28,21 @@ const (
 	defaultMaxRetry        = 3
 	collectorTLS           = gRPCPrefix + "tls"
 	collectorTLSCA         = gRPCPrefix + "tls.ca"
+	agentCert              = gRPCPrefix + "tls.cert"
+	agentKey               = gRPCPrefix + "tls.key"
 	collectorTLSServerName = gRPCPrefix + "tls.server-name"
 	discoveryMinPeers      = gRPCPrefix + "discovery.min-peers"
 )
 
 // AddFlags adds flags for Options.
 func AddFlags(flags *flag.FlagSet) {
-	flags.String(collectorHostPort, "", "Comma-separated string representing host:port of a static list of collectors to connect to directly.")
-	flags.Uint(retry, defaultMaxRetry, "Sets the maximum number of retries for a call.")
-	flags.Bool(collectorTLS, false, "Enable TLS.")
-	flags.String(collectorTLSCA, "", "Path to a TLS CA file. (default use the systems truststore)")
-	flags.String(collectorTLSServerName, "", "Override the TLS server name.")
+	flags.String(collectorHostPort, "", "Comma-separated string representing host:port of a static list of collectors to connect to directly")
+	flags.Uint(retry, defaultMaxRetry, "Sets the maximum number of retries for a call")
+	flags.Bool(collectorTLS, false, "Use TLS when talking to the remote collector")
+	flags.String(collectorTLSCA, "", "Path to a TLS CA file used to verify the remote server. (default use the systems truststore)")
+	flags.String(collectorTLSServerName, "", "Override the TLS server name we expected in the remote certificate")
+	flags.String(agentCert, "", "Path to a TLS client certificate file, used to identify this agent to the collector")
+	flags.String(agentKey, "", "Path to the TLS client key for the client certificate")
 	flags.Int(discoveryMinPeers, 3, "Max number of collectors to which the agent will try to connect at any given time")
 }
 
@@ -52,6 +56,8 @@ func (b *ConnBuilder) InitFromViper(v *viper.Viper) *ConnBuilder {
 	b.TLS = v.GetBool(collectorTLS)
 	b.TLSCA = v.GetString(collectorTLSCA)
 	b.TLSServerName = v.GetString(collectorTLSServerName)
+	b.TLSCert = v.GetString(agentCert)
+	b.TLSKey = v.GetString(agentKey)
 	b.DiscoveryMinPeers = v.GetInt(discoveryMinPeers)
 	return b
 }