feat(dgraph): making all internal communications with tls configured

.gitignore

@@ -31,3 +31,5 @@ dgraph.iml
 
 #darwin
 .DS_Store
+
+vendor

conn/node.go

@@ -19,6 +19,7 @@ package conn
 import (
 	"bytes"
 	"context"
+	"crypto/tls"
 	"encoding/binary"
 	"fmt"
 	"math/rand"
@@ -66,16 +67,17 @@ type Node struct {
 	_raft      raft.Node
 
 	// Fields which are never changed after init.
-	StartTime   time.Time
-	Cfg         *raft.Config
-	MyAddr      string
-	Id          uint64
-	peers       map[uint64]string
-	confChanges map[uint64]chan error
-	messages    chan sendmsg
-	RaftContext *pb.RaftContext
-	Store       *raftwal.DiskStorage
-	Rand        *rand.Rand
+	StartTime       time.Time
+	Cfg             *raft.Config
+	MyAddr          string
+	Id              uint64
+	peers           map[uint64]string
+	confChanges     map[uint64]chan error
+	messages        chan sendmsg
+	RaftContext     *pb.RaftContext
+	Store           *raftwal.DiskStorage
+	Rand            *rand.Rand
+	tlsClientConfig *tls.Config
 
 	Proposals proposals
 
@@ -84,7 +86,7 @@ type Node struct {
 }
 
 // NewNode returns a new Node instance.
-func NewNode(rc *pb.RaftContext, store *raftwal.DiskStorage) *Node {
+func NewNode(rc *pb.RaftContext, store *raftwal.DiskStorage, tlsConfig *tls.Config) *Node {
 	snap, err := store.Snapshot()
 	x.Check(err)
 
@@ -135,13 +137,14 @@ func NewNode(rc *pb.RaftContext, store *raftwal.DiskStorage) *Node {
 		},
 		// processConfChange etc are not throttled so some extra delta, so that we don't
 		// block tick when applyCh is full
-		Applied:     y.WaterMark{Name: fmt.Sprintf("Applied watermark")},
-		RaftContext: rc,
-		Rand:        rand.New(&lockedSource{src: rand.NewSource(time.Now().UnixNano())}),
-		confChanges: make(map[uint64]chan error),
-		messages:    make(chan sendmsg, 100),
-		peers:       make(map[uint64]string),
-		requestCh:   make(chan linReadReq, 100),
+		Applied:         y.WaterMark{Name: fmt.Sprintf("Applied watermark")},
+		RaftContext:     rc,
+		Rand:            rand.New(&lockedSource{src: rand.NewSource(time.Now().UnixNano())}),
+		confChanges:     make(map[uint64]chan error),
+		messages:        make(chan sendmsg, 100),
+		peers:           make(map[uint64]string),
+		requestCh:       make(chan linReadReq, 100),
+		tlsClientConfig: tlsConfig,
 	}
 	n.Applied.Init(nil)
 	// This should match up to the Applied index set above.
@@ -525,7 +528,7 @@ func (n *Node) Connect(pid uint64, addr string) {
 		n.SetPeer(pid, addr)
 		return
 	}
-	GetPools().Connect(addr)
+	GetPools().Connect(addr, n.tlsClientConfig)
 	n.SetPeer(pid, addr)
 }
 

conn/node_test.go

@@ -71,7 +71,7 @@ func TestProposal(t *testing.T) {
 	defer store.Closer.SignalAndWait()
 
 	rc := &pb.RaftContext{Id: 1}
-	n := NewNode(rc, store)
+	n := NewNode(rc, store, nil)
 
 	peers := []raft.Peer{{ID: n.Id}}
 	n.SetRaft(raft.StartNode(n.Cfg, peers))

conn/pool.go

@@ -18,6 +18,7 @@ package conn
 
 import (
 	"context"
+	"crypto/tls"
 	"sync"
 	"time"
 
@@ -30,6 +31,7 @@ import (
 	"go.opencensus.io/plugin/ocgrpc"
 
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
 )
 
 var (
@@ -138,13 +140,13 @@ func (p *Pools) getPool(addr string) (*Pool, bool) {
 }
 
 // Connect creates a Pool instance for the node with the given address or returns the existing one.
-func (p *Pools) Connect(addr string) *Pool {
+func (p *Pools) Connect(addr string, tlsClientConf *tls.Config) *Pool {
 	existingPool, has := p.getPool(addr)
 	if has {
 		return existingPool
 	}
 
-	pool, err := newPool(addr)
+	pool, err := newPool(addr, tlsClientConf)
 	if err != nil {
 		glog.Errorf("Unable to connect to host: %s", addr)
 		return nil
@@ -160,21 +162,32 @@ func (p *Pools) Connect(addr string) *Pool {
 	glog.Infof("CONNECTING to %s\n", addr)
 	p.all[addr] = pool
 	return pool
+
 }
 
 // newPool creates a new "pool" with one gRPC connection, refcount 0.
-func newPool(addr string) (*Pool, error) {
-	conn, err := grpc.Dial(addr,
+func newPool(addr string, tlsClientConf *tls.Config) (*Pool, error) {
+	conOpts := []grpc.DialOption {
 		grpc.WithStatsHandler(&ocgrpc.ClientHandler{}),
 		grpc.WithDefaultCallOptions(
 			grpc.MaxCallRecvMsgSize(x.GrpcMaxSize),
 			grpc.MaxCallSendMsgSize(x.GrpcMaxSize),
 			grpc.UseCompressor((snappyCompressor{}).Name())),
 		grpc.WithBackoffMaxDelay(time.Second),
-		grpc.WithInsecure())
+	}
+
+	if tlsClientConf != nil {
+		conOpts = append(conOpts, grpc.WithTransportCredentials(credentials.NewTLS(tlsClientConf)))
+	} else {
+		conOpts = append(conOpts, grpc.WithInsecure())
+	}
+
+	conn, err := grpc.Dial(addr, conOpts...)
 	if err != nil {
+		glog.Errorf("unable to connect with %s : %s", addr, err)
 		return nil, err
 	}
+
 	pl := &Pool{conn: conn, Addr: addr, lastEcho: time.Now(), closer: z.NewCloser(1)}
 	go pl.MonitorHealth()
 	return pl, nil

dgraph/cmd/alpha/run.go

@@ -63,11 +63,6 @@ import (
 	_ "github.com/vektah/gqlparser/v2/validator/rules" // make gql validator init() all rules
 )
 
-const (
-	tlsNodeCert = "node.crt"
-	tlsNodeKey  = "node.key"
-)
-
 var (
 	bindall bool
 
@@ -197,6 +192,11 @@ they form a Raft group and provide synchronous replication.
 	flag.String("tls_dir", "", "Path to directory that has TLS certificates and keys.")
 	flag.Bool("tls_use_system_ca", true, "Include System CA into CA Certs.")
 	flag.String("tls_client_auth", "VERIFYIFGIVEN", "Enable TLS client authentication")
+	flag.Bool("tls_internal_port_enabled", false, "(optional) enable inter node TLS encryption between cluster nodes.")
+	flag.String("tls_cert", "", "(optional) The Cert file name in tls_dir which is needed to "+
+		"connect as a client with the other nodes in the cluster.")
+	flag.String("tls_key", "", "(optional) The private key file name "+
+		"in tls_dir needed to connect as a client with the other nodes in the cluster.")
 
 	//Custom plugins.
 	flag.String("custom_tokenizers", "",
@@ -451,7 +451,7 @@ func setupServer(closer *z.Closer) {
 		laddr = "0.0.0.0"
 	}
 
-	tlsCfg, err := x.LoadServerTLSConfig(Alpha.Conf, tlsNodeCert, tlsNodeKey)
+	tlsCfg, err := x.LoadServerTLSConfig(Alpha.Conf, x.TLSNodeCert, x.TLSNodeKey)
 	if err != nil {
 		log.Fatalf("Failed to setup TLS: %v\n", err)
 	}
@@ -695,6 +695,8 @@ func run() {
 	abortDur, err := time.ParseDuration(Alpha.Conf.GetString("abort_older_than"))
 	x.Check(err)
 
+	tlsConf, err := x.LoadClientTLSConfigForInternalPort(Alpha.Conf)
+	x.Check(err)
 	x.WorkerConfig = x.WorkerOptions{
 		ExportPath:          Alpha.Conf.GetString("export"),
 		NumPendingProposals: Alpha.Conf.GetInt("pending_proposals"),
@@ -710,6 +712,9 @@ func run() {
 		AbortOlderThan:      abortDur,
 		StartTime:           startTime,
 		LudicrousMode:       Alpha.Conf.GetBool("ludicrous_mode"),
+		TLSClientConfig:     tlsConf,
+		TLSDir:              Alpha.Conf.GetString("tls_dir"),
+		TLSInterNodeEnabled: Alpha.Conf.GetBool("tls_internal_port_enabled"),
 	}
 	if x.WorkerConfig.EncryptionKey, err = enc.ReadKey(Alpha.Conf); err != nil {
 		glog.Infof("unable to read key %v", err)

dgraph/cmd/bulk/loader.go

@@ -21,6 +21,7 @@ import (
 	"compress/gzip"
 	"context"
 	"fmt"
+	"google.golang.org/grpc/credentials"
 	"hash/adler32"
 	"io"
 	"io/ioutil"
@@ -106,13 +107,20 @@ func newLoader(opt *options) *loader {
 	}
 
 	fmt.Printf("Connecting to zero at %s\n", opt.ZeroAddr)
-
 	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
 	defer cancel()
 
-	zero, err := grpc.DialContext(ctx, opt.ZeroAddr,
+	tlsConf, err := x.LoadClientTLSConfigForInternalPort(Bulk.Conf)
+	x.Check(err)
+	dialOpts := []grpc.DialOption{
 		grpc.WithBlock(),
-		grpc.WithInsecure())
+	}
+	if tlsConf != nil {
+		dialOpts = append(dialOpts, grpc.WithTransportCredentials(credentials.NewTLS(tlsConf)))
+	} else {
+		dialOpts = append(dialOpts, grpc.WithInsecure())
+	}
+	zero, err := grpc.DialContext(ctx, opt.ZeroAddr, dialOpts...)
 	x.Checkf(err, "Unable to connect to zero, Is it running at %s?", opt.ZeroAddr)
 	st := &state{
 		opt:    opt,

dgraph/cmd/bulk/run.go

@@ -108,7 +108,7 @@ func init() {
 	flag.String("badger.cache_percentage", "0,100",
 		"Cache percentages summing up to 100 for various caches"+
 			" (FORMAT: BlockCacheSize, IndexCacheSize).")
-
+	x.RegisterClientTLSFlags(flag)
 	// Encryption and Vault options
 	enc.RegisterFlags(flag)
 }

dgraph/cmd/live/run.go

@@ -35,6 +35,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/spf13/viper"
+	"google.golang.org/grpc"
+
 	"google.golang.org/grpc/metadata"
 
 	"github.com/dgraph-io/badger/v2"
@@ -363,7 +366,7 @@ func (l *loader) processLoadFile(ctx context.Context, rd *bufio.Reader, ck chunk
 	return nil
 }
 
-func setup(opts batchMutationOptions, dc *dgo.Dgraph) *loader {
+func setup(opts batchMutationOptions, dc *dgo.Dgraph, conf *viper.Viper) *loader {
 	var db *badger.DB
 	if len(opt.clientDir) > 0 {
 		x.Check(os.MkdirAll(opt.clientDir, 0700))
@@ -379,8 +382,12 @@ func setup(opts batchMutationOptions, dc *dgo.Dgraph) *loader {
 
 	}
 
+	dialOpts := []grpc.DialOption{}
+	tlsConfig, tlsErr := x.LoadClientTLSConfigForInternalPort(conf)
+	x.Check(tlsErr)
+
 	// compression with zero server actually makes things worse
-	connzero, err := x.SetupConnection(opt.zero, nil, false)
+	connzero, err := x.SetupConnection(opt.zero, tlsConfig, false, dialOpts...)
 	x.Checkf(err, "Unable to connect to zero, Is it running at %s?", opt.zero)
 
 	alloc := xidmap.New(connzero, db)
@@ -445,7 +452,7 @@ func run() error {
 	dg, closeFunc := x.GetDgraphClient(Live.Conf, true)
 	defer closeFunc()
 
-	l := setup(bmOpts, dg)
+	l := setup(bmOpts, dg, Live.Conf)
 	defer l.zeroconn.Close()
 
 	if len(opt.schemaFile) > 0 {

dgraph/cmd/zero/raft.go

@@ -209,7 +209,7 @@ func (n *node) handleMemberProposal(member *pb.Member) error {
 	}
 
 	// Create a connection to this server.
-	go conn.GetPools().Connect(member.Addr)
+	go conn.GetPools().Connect(member.Addr, n.server.tlsClientConfig)
 
 	group.Members[member.Id] = member
 	// Increment nextGroup when we have enough replicas
@@ -537,7 +537,7 @@ func (n *node) initAndStartNode() error {
 		}
 
 	case len(opts.peer) > 0:
-		p := conn.GetPools().Connect(opts.peer)
+		p := conn.GetPools().Connect(opts.peer, opts.tlsClientConfig)
 		if p == nil {
 			return errors.Errorf("Unhealthy connection to %v", opts.peer)
 		}