Secure Alter endpoint

Alpha can now set an auth token via commandline flag, which would be
checked for every incoming request to alter the DB. Only if the token
provided by the user matches against the one set in the flag, would the
Alter operation be allowed to run. This system provides a basic way to
avoid accidental schema changes.

All raft logs now use glog library to output.

conn/node.go

@@ -21,6 +21,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"log"
 	"math/rand"
 	"sync"
 	"time"
@@ -71,6 +72,22 @@ type Node struct {
 	Applied x.WaterMark
 }
 
+type raftLogger struct {
+}
+
+func (rl *raftLogger) Debug(v ...interface{})                   { glog.V(1).Info(v...) }
+func (rl *raftLogger) Debugf(format string, v ...interface{})   { glog.V(1).Infof(format, v...) }
+func (rl *raftLogger) Error(v ...interface{})                   { glog.Error(v...) }
+func (rl *raftLogger) Errorf(format string, v ...interface{})   { glog.Errorf(format, v...) }
+func (rl *raftLogger) Info(v ...interface{})                    { glog.Info(v...) }
+func (rl *raftLogger) Infof(format string, v ...interface{})    { glog.Infof(format, v...) }
+func (rl *raftLogger) Warning(v ...interface{})                 { glog.Warning(v...) }
+func (rl *raftLogger) Warningf(format string, v ...interface{}) { glog.Warningf(format, v...) }
+func (rl *raftLogger) Fatal(v ...interface{})                   { glog.Fatal(v...) }
+func (rl *raftLogger) Fatalf(format string, v ...interface{})   { glog.Fatalf(format, v...) }
+func (rl *raftLogger) Panic(v ...interface{})                   { log.Panic(v...) }
+func (rl *raftLogger) Panicf(format string, v ...interface{})   { log.Panicf(format, v...) }
+
 func NewNode(rc *pb.RaftContext, store *raftwal.DiskStorage) *Node {
 	snap, err := store.Snapshot()
 	x.Check(err)
@@ -86,7 +103,6 @@ func NewNode(rc *pb.RaftContext, store *raftwal.DiskStorage) *Node {
 			Storage:         store,
 			MaxSizePerMsg:   256 << 10,
 			MaxInflightMsgs: 256,
-			Logger:          &raft.DefaultLogger{Logger: x.Logger},
 			// We don't need lease based reads. They cause issues because they
 			// require CheckQuorum to be true, and that causes a lot of issues
 			// for us during cluster bootstrapping and later. A seemingly
@@ -116,6 +132,8 @@ func NewNode(rc *pb.RaftContext, store *raftwal.DiskStorage) *Node {
 			// like somehow the first index can be out of sync with the latest
 			// snapshot.
 			Applied: snap.Metadata.Index,
+
+			Logger: &raftLogger{},
 		},
 		// processConfChange etc are not throttled so some extra delta, so that we don't
 		// block tick when applyCh is full

contrib/integration/increment/main_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/dgraph-io/dgraph/x"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
 )
 
 const N = 10
@@ -101,7 +102,13 @@ func TestIncrement(t *testing.T) {
 	dg := dgo.NewDgraphClient(dc)
 
 	op := api.Operation{DropAll: true}
-	x.Check(dg.Alter(context.Background(), &op))
+
+	// The following piece of code shows how one can set metadata with
+	// auth-token, to allow Alter operation, if the server requires it.
+	md := metadata.New(nil)
+	md.Append("auth-token", "mrjn2")
+	ctx := metadata.NewOutgoingContext(context.Background(), md)
+	x.Check(dg.Alter(ctx, &op))
 
 	cnt, err := process(dg, false)
 	if err != nil {

dgraph/cmd/alpha/http.go

@@ -33,6 +33,8 @@ import (
 	"github.com/dgraph-io/dgraph/query"
 	"github.com/dgraph-io/dgraph/worker"
 	"github.com/dgraph-io/dgraph/x"
+	"github.com/golang/glog"
+	"google.golang.org/grpc/metadata"
 )
 
 func allowed(method string) bool {
@@ -394,8 +396,17 @@ func alterHandler(w http.ResponseWriter, r *http.Request) {
 		op.Schema = string(b)
 	}
 
-	_, err = (&edgraph.Server{}).Alter(context.Background(), op)
-	if err != nil {
+	glog.Infof("Got alter request via HTTP from %s\n", r.RemoteAddr)
+	fwd := r.Header.Get("X-Forwarded-For")
+	if len(fwd) > 0 {
+		glog.Infof("The alter request is forwarded by %s\n", fwd)
+	}
+
+	md := metadata.New(nil)
+	// Pass in an auth token, if present.
+	md.Append("auth-token", r.Header.Get("X-Dgraph-AuthToken"))
+	ctx := metadata.NewIncomingContext(context.Background(), md)
+	if _, err = (&edgraph.Server{}).Alter(ctx, op); err != nil {
 		x.SetStatus(w, x.Error, err.Error())
 		return
 	}

dgraph/cmd/alpha/run.go

@@ -104,6 +104,10 @@ they form a Raft group and provide synchronous replication.
 		"Enables the expand() feature. This is very expensive for large data loads because it"+
 			" doubles the number of mutations going on in the system.")
 
+	flag.StringVar(&config.AuthToken, "auth_token", "",
+		"If set, all Alter requests to Dgraph would need to have this token."+
+			" The token can be passed as follows: For HTTP requests, in X-Dgraph-AuthToken header."+
+			" For Grpc, in auth-token key in the context.")
 	flag.Float64VarP(&config.AllottedMemory, "lru_mb", "l", -1,
 		"Estimated memory the LRU cache can take. "+
 			"Actual usage by the process would be more than specified here.")

dgraph/cmd/version/run.go

@@ -17,6 +17,7 @@
 package version
 
 import (
+	"fmt"
 	"os"
 
 	"github.com/spf13/cobra"
@@ -32,7 +33,7 @@ func init() {
 		Short: "Prints the dgraph version details",
 		Long:  "Version prints the dgraph version as reported by the build details.",
 		Run: func(cmd *cobra.Command, args []string) {
-			x.PrintVersion()
+			fmt.Print(x.BuildDetails())
 			os.Exit(0)
 		},
 	}

edgraph/config.go

@@ -36,6 +36,7 @@ type Options struct {
 	BadgerVlog   string
 	WALDir       string
 	Nomutations  bool
+	AuthToken    string
 
 	AllottedMemory float64
 

edgraph/server.go

@@ -28,6 +28,8 @@ import (
 	"unicode"
 
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/status"
 
 	"golang.org/x/net/context"
@@ -267,26 +269,30 @@ func (s *ServerState) getTimestamp(readOnly bool) uint64 {
 func (s *Server) Alter(ctx context.Context, op *api.Operation) (*api.Payload, error) {
 	// Always print out Alter operations because they are important and rare.
 	glog.Infof("Received ALTER op: %+v", op)
-	defer glog.Infof("ALTER op: %+v done", op)
 
+	// The following code block checks if the operation should run or not.
 	if op.Schema == "" && op.DropAttr == "" && !op.DropAll {
 		// Must have at least one field set. This helps users if they attempt
 		// to set a field but use the wrong name (could be decoded from JSON).
 		return nil, x.Errorf("Operation must have at least one field set")
 	}
-
 	empty := &api.Payload{}
 	if err := x.HealthCheck(); err != nil {
 		if tr, ok := trace.FromContext(ctx); ok {
 			tr.LazyPrintf("Request rejected %v", err)
 		}
 		return empty, err
 	}
-
 	if !isMutationAllowed(ctx) {
-		return nil, x.Errorf("No mutations allowed.")
+		return nil, x.Errorf("No mutations allowed by server.")
 	}
+	if err := isAlterAllowed(ctx); err != nil {
+		glog.Warningf("Alter denied with error: %v\n", err)
+		return nil, err
+	}
+	// All checks done.
 
+	defer glog.Infof("ALTER op: %+v done", op)
 	// StartTs is not needed if the predicate to be dropped lies on this server but is required
 	// if it lies on some other machine. Let's get it for safety.
 	m := &pb.Mutations{StartTs: State.getTimestamp(false)}
@@ -315,7 +321,7 @@ func (s *Server) Alter(ctx context.Context, op *api.Operation) (*api.Payload, er
 	if err != nil {
 		return empty, err
 	}
-	x.Printf("Got schema: %+v\n", updates)
+	glog.Infof("Got schema: %+v\n", updates)
 	// TODO: Maybe add some checks about the schema.
 	m.Schema = updates
 	_, err = query.ApplyMutations(ctx, m)
@@ -568,6 +574,30 @@ func isMutationAllowed(ctx context.Context) bool {
 	return true
 }
 
+var errNoAuth = x.Errorf("No Auth Token found. Token needed for Alter operations.")
+
+func isAlterAllowed(ctx context.Context) error {
+	p, ok := peer.FromContext(ctx)
+	if ok {
+		glog.Infof("Got Alter request from %q\n", p.Addr)
+	}
+	if len(Config.AuthToken) == 0 {
+		return nil
+	}
+	md, ok := metadata.FromIncomingContext(ctx)
+	if !ok {
+		return errNoAuth
+	}
+	tokens := md.Get("auth-token")
+	if len(tokens) == 0 {
+		return errNoAuth
+	}
+	if tokens[0] != Config.AuthToken {
+		return x.Errorf("Provided auth token [%s] does not match. Permission denied.", tokens[0])
+	}
+	return nil
+}
+
 func parseNQuads(b []byte) ([]*api.NQuad, error) {
 	var nqs []*api.NQuad
 	for _, line := range bytes.Split(b, []byte{'\n'}) {

worker/mutation.go

@@ -36,6 +36,7 @@ import (
 	"github.com/dgraph-io/dgraph/schema"
 	"github.com/dgraph-io/dgraph/types"
 	"github.com/dgraph-io/dgraph/x"
+	"github.com/golang/glog"
 )
 
 var (
@@ -126,7 +127,7 @@ func runSchemaMutationHelper(ctx context.Context, update *pb.SchemaUpdate, start
 	// index mutations (old set and new del)
 	// We need watermark for index/reverse edge addition for linearizable reads.
 	// (both applied and synced watermarks).
-	defer x.Printf("Done schema update %+v\n", update)
+	defer glog.Infof("Done schema update %+v\n", update)
 	if !ok {
 		if current.Directive == pb.SchemaUpdate_INDEX {
 			if err := n.rebuildOrDelIndex(ctx, update.Predicate, true, startTs); err != nil {

x/init.go

@@ -18,6 +18,8 @@ package x
 
 import (
 	"fmt"
+
+	"github.com/golang/glog"
 )
 
 var (
@@ -75,7 +77,7 @@ Licensed under Apache 2.0. Copyright 2015-2018 Dgraph Labs, Inc.
 
 // PrintVersionOnly prints version and other helpful information if --version.
 func PrintVersion() {
-	fmt.Println(BuildDetails())
+	glog.Infof("\n%s\n", BuildDetails())
 }
 
 func Version() string {