Secure communication

cmd/dkvctl/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"fmt"
+	utils "github.com/flipkart-incubator/dkv/internal"
 	"os"
 	"sort"
 	"strings"
@@ -228,11 +229,21 @@ func (c *cmd) clusterInfo(client *ctl.DKVClient, args ...string) {
 	}
 }
 
-var dkvAddr, dkvAuthority string
+var (
+	dkvAddr      string
+	caCertPath   string
+	dkvAuthority string
+)
 
 func init() {
+<<<<<<< HEAD
 	flag.StringVarP(&dkvAddr, "dkvAddr", "a", "127.0.0.1:8080", "<host>:<port> - DKV server address")
+=======
+	flag.StringVar(&dkvAddr, "dkvAddr", "127.0.0.1:8080", "<host>:<port> - DKV server address")
+	flag.StringVar(&caCertPath, "caCertPath", "", "Path for root certificate of the chain, i.e. CA certificate")
+>>>>>>> 5cda07c6b9de1771e120145aad6d107a9eb3f108
 	flag.StringVar(&dkvAuthority, "authority", "", "Override :authority pseudo header for routing purposes. Useful while accessing DKV via service mesh.")
+
 	for _, c := range cmds {
 		if c.argDesc == "" {
 			flag.BoolVar(&c.emptyValue, c.name, c.emptyValue, c.cmdDesc)
@@ -247,6 +258,7 @@ func init() {
 	flag.CommandLine.SortFlags = false
 }
 
+<<<<<<< HEAD
 func isFlagPassed(name string) bool {
 	found := false
 	flag.Visit(func(f *flag.Flag) {
@@ -255,6 +267,27 @@ func isFlagPassed(name string) bool {
 		}
 	})
 	return found
+=======
+func usage() {
+	fmt.Printf("Usage of %s:\n", os.Args[0])
+	printUsage([]string{"dkvAddr", "authority", "caCertPath"})
+	for _, cmd := range cmds {
+		cmd.usage()
+	}
+}
+
+func printUsage(flags []string) {
+	for _, flagName := range flags {
+		dkvFlag := flag.Lookup(flagName)
+		if dkvFlag != nil {
+			fmt.Printf("  -%s %s (default: %s)\n", dkvFlag.Name, dkvFlag.Usage, dkvFlag.DefValue)
+		}
+	}
+}
+
+func trimLower(str string) string {
+	return strings.ToLower(strings.TrimSpace(str))
+>>>>>>> 5cda07c6b9de1771e120145aad6d107a9eb3f108
 }
 
 func main() {
@@ -269,7 +302,12 @@ func main() {
 		fmt.Printf(" (:authority = %s)", dkvAuthority)
 	}
 	fmt.Printf("...")
+<<<<<<< HEAD
 	client, err := ctl.NewInSecureDKVClient(dkvAddr, dkvAuthority, ctl.DefaultConnectOpts)
+=======
+	client, err := utils.NewDKVClient(utils.DKVConfig{
+		SrvrAddr: dkvAddr, CaCertPath: caCertPath}, dkvAuthority)
+>>>>>>> 5cda07c6b9de1771e120145aad6d107a9eb3f108
 	if err != nil {
 		fmt.Printf("\nUnable to create DKV client. Error: %v\n", err)
 		return
@@ -293,4 +331,4 @@ func main() {
 	if !validCmd {
 		flag.Usage()
 	}
-}
+}

cmd/dkvsrv/main.go

@@ -6,16 +6,23 @@ import (
 	"io"
 	"io/ioutil"
 	"log"
+<<<<<<< HEAD
 	"net"
 	"net/http"
 	"net/url"
+=======
+>>>>>>> 5cda07c6b9de1771e120145aad6d107a9eb3f108
 	"os"
 	"os/signal"
 	"path"
 	"strings"
 	"syscall"
 
+<<<<<<< HEAD
 	"github.com/flipkart-incubator/dkv/internal/discovery"
+=======
+	utils "github.com/flipkart-incubator/dkv/internal"
+>>>>>>> 5cda07c6b9de1771e120145aad6d107a9eb3f108
 	"github.com/flipkart-incubator/dkv/internal/master"
 	"github.com/flipkart-incubator/dkv/internal/opts"
 	"github.com/flipkart-incubator/dkv/internal/slave"
@@ -25,6 +32,7 @@ import (
 	"github.com/flipkart-incubator/dkv/internal/storage/badger"
 	"github.com/flipkart-incubator/dkv/internal/storage/rocksdb"
 	"github.com/flipkart-incubator/dkv/internal/sync"
+<<<<<<< HEAD
 	"github.com/flipkart-incubator/dkv/pkg/health"
 	"github.com/flipkart-incubator/dkv/pkg/serverpb"
 	nexus_api "github.com/flipkart-incubator/nexus/pkg/api"
@@ -65,6 +73,30 @@ var (
 
 	//nexus flags
 	nexusLogDirFlag, nexusSnapDirFlag *flag.Flag
+=======
+	"github.com/flipkart-incubator/dkv/pkg/serverpb"
+	nexus_api "github.com/flipkart-incubator/nexus/pkg/api"
+	nexus "github.com/flipkart-incubator/nexus/pkg/raft"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+)
+
+var (
+	disklessMode     bool
+	dbEngine         string
+	dbEngineIni      string
+	dbFolder         string
+	dbListenAddr     string
+	peerListenAddr   string
+	dbRole           string
+	statsdAddr       string
+	replMasterAddr   string
+	replPollInterval time.Duration
+	certPath         string
+	keyPath          string
+	caCertPath       string
+	blockCacheSize   uint64
+>>>>>>> 5cda07c6b9de1771e120145aad6d107a9eb3f108
 
 	// Logging vars
 	verboseLogging bool
@@ -82,7 +114,27 @@ var (
 )
 
 func init() {
+<<<<<<< HEAD
 	initializeFlags()
+=======
+	flag.BoolVar(&disklessMode, "dbDiskless", false, fmt.Sprintf("Enables diskless mode where data is stored entirely in memory.\nAvailable on Badger for standalone and slave roles. (default %v)", disklessMode))
+	flag.StringVar(&dbFolder, "dbFolder", "/tmp/dkvsrv", "DB folder path for storing data files")
+	flag.StringVar(&dbListenAddr, "dbListenAddr", "127.0.0.1:8080", "Address on which the DKV service binds")
+	flag.StringVar(&peerListenAddr, "peerListenAddr", "127.0.0.1:8083", "Address on which the DKV replication service binds")
+	flag.StringVar(&dbEngine, "dbEngine", "rocksdb", "Underlying DB engine for storing data - badger|rocksdb")
+	flag.StringVar(&dbEngineIni, "dbEngineIni", "", "An .ini file for configuring the underlying storage engine. Refer badger.ini or rocks.ini for more details.")
+	flag.StringVar(&dbRole, "dbRole", "none", "DB role of this node - none|master|slave")
+	flag.StringVar(&statsdAddr, "statsdAddr", "", "StatsD service address in host:port format")
+	flag.StringVar(&replMasterAddr, "replMasterAddr", "", "Service address of DKV master node for replication")
+	flag.DurationVar(&replPollInterval, "replPollInterval", 5*time.Second, "Interval used for polling changes from master. Eg., 10s, 5ms, 2h, etc.")
+	flag.StringVar(&dbAccessLog, "dbAccessLog", "", "File for logging DKV accesses eg., stdout, stderr, /tmp/access.log")
+	flag.StringVar(&certPath, "certPath", "", "Path for certificate file of this node")
+	flag.StringVar(&keyPath, "keyPath", "", "Path for key file of this node")
+	flag.StringVar(&caCertPath, "caCertPath", "", "Path for root certificate of the chain, i.e. CA certificate")
+	flag.BoolVar(&verboseLogging, "verbose", false, fmt.Sprintf("Enable verbose logging.\nBy default, only warnings and errors are logged. (default %v)", verboseLogging))
+	flag.Uint64Var(&blockCacheSize, "block-cache-size", defBlockCacheSize, "Amount of cache (in bytes) to set aside for data blocks. A value of 0 disables block caching altogether.")
+	setDKVDefaultsForNexusDirs()
+>>>>>>> 5cda07c6b9de1771e120145aad6d107a9eb3f108
 }
 
 func initializeFlags() {
@@ -105,8 +157,19 @@ func main() {
 	setupStats()
 	go setupHttpServer()
 
+	var secure = caCertPath != "" && keyPath != "" && certPath != ""
+
+	var srvrMode utils.ConnectionMode
+	if secure {
+		srvrMode = utils.ServerTLS
+	} else {
+		srvrMode = utils.Insecure
+	}
+
 	kvs, cp, ca, br := newKVStore()
-	grpcSrvr, lstnr := newGrpcServerListener()
+	grpcSrvr, lstnr := utils.NewGrpcServerListener(utils.DKVConfig{ConnectionMode: srvrMode,
+		SrvrAddr: dbListenAddr, KeyPath: keyPath, CertPath: certPath,
+		CaCertPath: caCertPath}, accessLogger)
 	defer grpcSrvr.GracefulStop()
 	srvrRole := toDKVSrvrRole(config.DbRole)
 	//srvrRole.printFlags()
@@ -173,6 +236,7 @@ func main() {
 		}
 		defer dkvSvc.Close()
 		serverpb.RegisterDKVServer(grpcSrvr, dkvSvc)
+<<<<<<< HEAD
 		serverpb.RegisterDKVReplicationServer(grpcSrvr, dkvSvc)
 		health.RegisterHealthServer(grpcSrvr, dkvSvc)
 
@@ -182,6 +246,35 @@ func main() {
 			if err != nil {
 				log.Panicf("Failed to start Discovery Service %v.", err)
 			}
+=======
+		var replSrvrMode utils.ConnectionMode
+		if secure {
+			replSrvrMode = utils.MutualTLS
+		} else {
+			replSrvrMode = utils.Insecure
+		}
+		replGrpcSrvr, replLstnr := utils.NewGrpcServerListener(utils.DKVConfig{ConnectionMode: replSrvrMode,
+			SrvrAddr: peerListenAddr, KeyPath: keyPath, CertPath: certPath,
+			CaCertPath: caCertPath}, accessLogger)
+		defer replGrpcSrvr.GracefulStop()
+		serverpb.RegisterDKVReplicationServer(replGrpcSrvr, dkvSvc)
+		go replGrpcSrvr.Serve(replLstnr)
+	case slaveRole:
+		var replClientMode utils.ConnectionMode
+		if secure {
+			replClientMode = utils.MutualTLS
+		} else {
+			replClientMode = utils.Insecure
+		}
+
+		// TODO: Check if authority override option is needed for slaves
+		// while they connect with masters
+		if replCli, err := utils.NewDKVClient(utils.DKVConfig{ConnectionMode: replClientMode,
+			SrvrAddr: replMasterAddr, KeyPath: keyPath, CertPath: certPath,
+			CaCertPath: caCertPath}, "");
+			err != nil {
+			panic(err)
+>>>>>>> 5cda07c6b9de1771e120145aad6d107a9eb3f108
 		} else {
 			// Currently nodes can be either discovery server or client. This will change when a node supports multiple regions
 			discoveryClient.RegisterRegion(dkvSvc)
@@ -209,6 +302,42 @@ func main() {
 	log.Printf("[WARN] Caught signal: %v. Shutting down...\n", sig)
 }
 
+<<<<<<< HEAD
+=======
+func validateFlags() {
+	if dbListenAddr != "" && strings.IndexRune(dbListenAddr, ':') < 0 {
+		log.Panicf("given listen address: %s is invalid, must be in host:port format", dbListenAddr)
+	}
+	if peerListenAddr != "" && strings.IndexRune(peerListenAddr, ':') < 0 {
+		log.Panicf("given listen address: %s is invalid, must be in host:port format", dbListenAddr)
+	}
+	if replMasterAddr != "" && strings.IndexRune(replMasterAddr, ':') < 0 {
+		log.Panicf("given master address: %s for replication is invalid, must be in host:port format", replMasterAddr)
+	}
+	if statsdAddr != "" && strings.IndexRune(statsdAddr, ':') < 0 {
+		log.Panicf("given StatsD address: %s is invalid, must be in host:port format", statsdAddr)
+	}
+
+	if disklessMode && strings.ToLower(dbEngine) == "rocksdb" {
+		log.Panicf("diskless is available only on Badger storage")
+	}
+
+	if strings.ToLower(dbRole) == slaveRole && replMasterAddr == "" {
+		log.Panicf("repl-master-addr must be given in slave mode")
+	}
+	nonNullAuthFlags := btou(certPath != "", keyPath != "", caCertPath != "")
+	if nonNullAuthFlags > 0 && nonNullAuthFlags < 3 {
+		log.Panicf("Missing TLS attributes, set all flags (caCertPath, keyPath, certPath) to run DKV in secure mode")
+	}
+
+	if dbEngineIni != "" {
+		if _, err := os.Stat(dbEngineIni); err != nil && os.IsNotExist(err) {
+			log.Panicf("given storage configuration file: %s does not exist", dbEngineIni)
+		}
+	}
+}
+
+>>>>>>> 5cda07c6b9de1771e120145aad6d107a9eb3f108
 func setupAccessLogger() {
 	accessLogger = zap.NewNop()
 	if config.AccessLog != "" {
@@ -284,6 +413,7 @@ func setupDKVLogger() {
 	}
 }
 
+<<<<<<< HEAD
 func newGrpcServerListener() (*grpc.Server, net.Listener) {
 	grpcSrvr := grpc.NewServer(
 		grpc.StreamInterceptor(grpc_zap.StreamServerInterceptor(accessLogger)),
@@ -302,6 +432,8 @@ func newListener() (lis net.Listener) {
 	return
 }
 
+=======
+>>>>>>> 5cda07c6b9de1771e120145aad6d107a9eb3f108
 func setupSignalHandler() <-chan os.Signal {
 	signals := []os.Signal{syscall.SIGINT, syscall.SIGQUIT, syscall.SIGSTOP, syscall.SIGTERM}
 	stopChan := make(chan os.Signal, len(signals))
@@ -323,7 +455,26 @@ func toDKVSrvrRole(role string) dkvSrvrRole {
 	return dkvSrvrRole(strings.TrimSpace(strings.ToLower(role)))
 }
 
+<<<<<<< HEAD
 func setFlagsForNexusDirs() {
+=======
+func (role dkvSrvrRole) printFlags() {
+	log.Println("Launching DKV server with following flags:")
+	switch role {
+	case noRole:
+		printFlagsWithPrefix("db")
+	case masterRole:
+		if haveFlagsWithPrefix("nexus") {
+			printFlagsWithPrefix("db", "nexus", "peer")
+		} else {
+			printFlagsWithPrefix("db", "peer")
+		}
+	case slaveRole:
+		printFlagsWithPrefix("db", "repl")
+	}
+	printFlagsWithoutPrefix("db", "repl", "nexus")
+}
+>>>>>>> 5cda07c6b9de1771e120145aad6d107a9eb3f108
 
 	nexusLogDirFlag, nexusSnapDirFlag = flag.Lookup("nexus-log-dir"), flag.Lookup("nexus-snap-dir")
 	if nexusLogDirFlag.Value.String() == "" {
@@ -426,6 +577,7 @@ func newDKVReplicator(kvs storage.KVStore) nexus_api.RaftReplicator {
 	}
 }
 
+<<<<<<< HEAD
 func registerDiscoveryServer(grpcSrvr *grpc.Server, dkvService master.DKVService) error {
 	discoveryService, err := discovery.NewDiscoveryService(dkvService, dkvLogger, &config.DiscoveryConfig.ServerConfig)
 	if err != nil {
@@ -585,4 +737,14 @@ func clusterMetricsHandler(w http.ResponseWriter, r *http.Request) {
 			}
 		}
 	}
+=======
+func btou(conditions... bool) int {
+	cnt := 0
+	for _, cond := range conditions {
+		if cond {
+			cnt += 1
+		}
+	}
+	return cnt
+>>>>>>> 5cda07c6b9de1771e120145aad6d107a9eb3f108
 }

go.mod

@@ -21,6 +21,7 @@ require (
 	github.com/prometheus/client_model v0.2.0
 	github.com/prometheus/procfs v0.0.10 // indirect
 	github.com/smira/go-statsd v1.3.1
+<<<<<<< HEAD
 	github.com/spf13/pflag v1.0.5
 	github.com/spf13/viper v1.10.1
 	github.com/vmihailenco/msgpack/v5 v5.3.4
@@ -29,6 +30,18 @@ require (
 	google.golang.org/grpc v1.43.0
 	google.golang.org/protobuf v1.27.1
 	gopkg.in/ini.v1 v1.66.2
+=======
+	go.uber.org/zap v1.14.1
+	golang.org/x/lint v0.0.0-20200302205851-738671d3881b // indirect
+	golang.org/x/net v0.0.0-20201002202402-0a1ea396d57c // indirect
+	golang.org/x/sys v0.0.0-20201005065044-765f4ea38db3 // indirect
+	golang.org/x/text v0.3.2 // indirect
+	golang.org/x/tools v0.0.0-20210108195828-e2f9c7f1fc8e // indirect
+	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
+	google.golang.org/grpc v1.28.0
+	google.golang.org/protobuf v1.26.0
+	gopkg.in/ini.v1 v1.62.0
+>>>>>>> 5cda07c6b9de1771e120145aad6d107a9eb3f108
 	honnef.co/go/tools v0.0.1-2020.1.3 // indirect
 )