Enterprise - Support reading a signed file with details of Enterprise license

dgraph/cmd/alpha/admin_backup.go

@@ -42,10 +42,9 @@ func backupHandler(w http.ResponseWriter, r *http.Request) {
 	if !handlerInit(w, r, http.MethodPost) {
 		return
 	}
-	if !Alpha.Conf.GetBool("enterprise_features") {
-		x.SetStatus(w,
-			"You must enable Dgraph enterprise features first. "+
-				"Restart Dgraph Alpha with --enterprise_features",
+	if !worker.EnterpriseEnabled() {
+		x.SetStatus(w, "You must enable enterprise features first. "+
+			"Supply the appropriate license file to Dgraph Zero using a flag or the HTTP endpoint.",
 			"Backup failed.")
 		return
 	}

dgraph/cmd/alpha/run.go

@@ -93,8 +93,6 @@ they form a Raft group and provide synchronous replication.
 	// with the flag name so that the values are picked up by Cobra/Viper's various config inputs
 	// (e.g, config file, env vars, cli flags, etc.)
 	flag := Alpha.Cmd.Flags()
-	flag.Bool("enterprise_features", false, "Enable Dgraph enterprise features. "+
-		"If you set this to true, you agree to the Dgraph Community License.")
 	flag.StringP("postings", "p", "p", "Directory to store posting lists.")
 
 	// Options around how to set up Badger.
@@ -435,11 +433,6 @@ func run() {
 
 	secretFile := Alpha.Conf.GetString("acl_secret_file")
 	if secretFile != "" {
-		if !Alpha.Conf.GetBool("enterprise_features") {
-			glog.Fatalf("You must enable Dgraph enterprise features with the " +
-				"--enterprise_features option in order to use ACL.")
-		}
-
 		hmacSecret, err := ioutil.ReadFile(secretFile)
 		if err != nil {
 			glog.Fatalf("Unable to read HMAC secret from file: %v", secretFile)

dgraph/cmd/zero/http.go

@@ -17,8 +17,10 @@
 package zero
 
 import (
+	"bytes"
 	"context"
 	"fmt"
+	"io/ioutil"
 	"net"
 	"net/http"
 	"strconv"
@@ -232,6 +234,35 @@ func (st *state) getState(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+func (st *state) applyEnterpriseLicense(w http.ResponseWriter, r *http.Request) {
+	x.AddCorsHeaders(w)
+	if r.Method == "OPTIONS" {
+		return
+	}
+	if r.Method != http.MethodPost {
+		w.WriteHeader(http.StatusBadRequest)
+		x.SetStatus(w, x.ErrorInvalidMethod, "Invalid method")
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	b, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		x.SetStatus(w, x.ErrorInvalidRequest, err.Error())
+		return
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+	if err := st.zero.applyEnterpriseLicense(ctx, bytes.NewReader(b)); err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		x.SetStatus(w, x.ErrorInvalidRequest, err.Error())
+		return
+	}
+	x.SetStatus(w, x.Success, "Done")
+}
+
 func (st *state) serveHTTP(l net.Listener) {
 	srv := &http.Server{
 		ReadTimeout:  10 * time.Second,

dgraph/cmd/zero/pgp.go

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2017-2018 Dgraph Labs, Inc. and Contributors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package zero
+
+import (
+	"encoding/json"
+	"io"
+	"io/ioutil"
+
+	"github.com/pkg/errors"
+	"golang.org/x/crypto/openpgp"
+	"golang.org/x/crypto/openpgp/armor"
+)
+
+func enterpriseDetails(signedFile, publicKey io.Reader, e *enterprise) error {
+	entityList, err := openpgp.ReadArmoredKeyRing(publicKey)
+	if err != nil {
+		return errors.Wrapf(err, "while reading public key")
+	}
+
+	// The signed file is expected to be have ASCII encoding, so we have to decode it before
+	// reading.
+	b, err := armor.Decode(signedFile)
+	if err != nil {
+		return errors.Wrapf(err, "while decoding license file")
+	}
+
+	md, err := openpgp.ReadMessage(b.Body, entityList, nil, nil)
+	if err != nil {
+		return errors.Wrapf(err, "while reading PGP message from license file")
+	}
+
+	// We need to read the body for the signature verification check to happen.
+	// md.Signature would be non-nil after reading the body if the verification is successfull.
+	buf, err := ioutil.ReadAll(md.UnverifiedBody)
+	if err != nil {
+		return errors.Wrapf(err, "while reading body from signed license file")
+	}
+	// This could be nil even if signature verification failed, so we also check Signature == nil
+	// below.
+	if md.SignatureError != nil {
+		return errors.Wrapf(md.SignatureError, "signature error while trying to verify license file")
+	}
+	if md.Signature == nil {
+		return errors.New("invalid signature while trying to verify license file")
+	}
+
+	err = json.Unmarshal(buf, e)
+	if err != nil {
+		return errors.Wrapf(err, "while JSON unmarshaling body of license file")
+	}
+	if e.Entity == "" || e.MaxNodes == 0 || e.Expiry.IsZero() {
+		return errors.Errorf("invalid JSON data, fields shouldn't be zero: %+v\n", e)
+	}
+	return nil
+}

dgraph/cmd/zero/pgp_test.go

@@ -0,0 +1,124 @@
+package zero
+
+import (
+	"bytes"
+	"crypto"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"golang.org/x/crypto/openpgp"
+	"golang.org/x/crypto/openpgp/armor"
+	"golang.org/x/crypto/openpgp/packet"
+)
+
+func encodePublicKey(t *testing.T, e *openpgp.Entity) *bytes.Buffer {
+	b := new(bytes.Buffer)
+	encodedKeyBuf, err := armor.Encode(b, openpgp.PublicKeyType, nil)
+	require.NoError(t, err)
+	err = e.Serialize(encodedKeyBuf)
+	require.NoError(t, err)
+	err = encodedKeyBuf.Close()
+	require.NoError(t, err)
+	return b
+}
+
+func signAndWriteMessage(t *testing.T, entity *openpgp.Entity, json string) *bytes.Buffer {
+	b := new(bytes.Buffer)
+	w, err := openpgp.Sign(b, entity, nil, &packet.Config{
+		RSABits:     4096,
+		DefaultHash: crypto.SHA512,
+	})
+	require.NoError(t, err)
+
+	_, err = w.Write([]byte(json))
+	require.NoError(t, err)
+
+	err = w.Close()
+	require.NoError(t, err)
+
+	// armor encode the message
+	abuf := new(bytes.Buffer)
+	w, err = armor.Encode(abuf, "PGP MESSAGE", nil)
+	_, err = w.Write(b.Bytes())
+	require.NoError(t, err)
+
+	err = w.Close()
+	require.NoError(t, err)
+
+	return abuf
+}
+
+func TestEnterpriseDetails(t *testing.T) {
+	correctEntity, err := openpgp.NewEntity("correct", "", "correct@correct.com", &packet.Config{
+		RSABits:     4096,
+		DefaultHash: crypto.SHA512,
+	})
+
+	require.NoError(t, err)
+	incorrectEntity, err := openpgp.NewEntity("incorrect", "", "incorrect@incorrect.com", &packet.Config{
+		RSABits:     4096,
+		DefaultHash: crypto.SHA512,
+	})
+	require.NoError(t, err)
+	correctJSON := `{"entity": "entity", "max_nodes": 10, "expiry": "2019-08-16T19:09:06+10:00"}`
+	correctTime, err := time.Parse(time.RFC3339, "2019-08-16T19:09:06+10:00")
+	require.NoError(t, err)
+
+	var tests = []struct {
+		name            string
+		signingEntity   *openpgp.Entity
+		json            string
+		verifyingEntity *openpgp.Entity
+		expectError     bool
+		expectedOutput  enterprise
+	}{
+		{
+			"Signing JSON with empty data should return an error",
+			correctEntity,
+			`{}`,
+			correctEntity,
+			true,
+			enterprise{},
+		},
+		{
+			"Signing JSON with incorrect private key should return an error",
+			incorrectEntity,
+			correctJSON,
+			correctEntity,
+			true,
+			enterprise{},
+		},
+		{
+			"Verifying data with incorrect public key should return an error",
+			correctEntity,
+			correctJSON,
+			incorrectEntity,
+			true,
+			enterprise{},
+		},
+		{
+			"Verifying data with correct public key should return correct data",
+			correctEntity,
+			correctJSON,
+			correctEntity,
+			false,
+			enterprise{"entity", 10, correctTime},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Logf("Running: %s\n", tt.name)
+		buf := signAndWriteMessage(t, tt.signingEntity, tt.json)
+		e := enterprise{}
+		publicKey := encodePublicKey(t, tt.verifyingEntity)
+		err = enterpriseDetails(buf, publicKey, &e)
+		if tt.expectError {
+			require.Error(t, err)
+			continue
+		}
+
+		require.NoError(t, err)
+		require.Equal(t, tt.expectedOutput, e)
+	}
+}

dgraph/cmd/zero/raft.go

@@ -75,6 +75,8 @@ func (n *node) uniqueKey() string {
 
 var errInternalRetry = errors.New("Retry Raft proposal internally")
 
+// proposeAndWait makes a proposal to the quorum for Group Zero and waits for it to be accepted by
+// the group before returning. It is safe to call concurrently.
 func (n *node) proposeAndWait(ctx context.Context, proposal *pb.ZeroProposal) error {
 	switch {
 	case n.Raft() == nil:
@@ -356,6 +358,18 @@ func (n *node) applyProposal(e raftpb.Entry) (string, error) {
 			return p.Key, err
 		}
 	}
+	if p.Enterprise != nil {
+		// Check that the number of nodes in the cluster should be less than MaxNodes, otherwise
+		// reject the proposal.
+		numNodes := len(state.GetZeros())
+		for _, group := range state.GetGroups() {
+			numNodes += len(group.GetMembers())
+		}
+		if uint64(numNodes) > p.GetEnterprise().GetMaxNodes() {
+			return p.Key, errInvalidProposal
+		}
+		state.Enterprise = p.Enterprise
+	}
 
 	if p.MaxLeaseId > state.MaxLeaseId {
 		state.MaxLeaseId = p.MaxLeaseId
@@ -494,6 +508,7 @@ func (n *node) initAndStartNode() error {
 					return
 				}
 				if err == errInvalidProposal {
+					glog.Errorf("invalid proposal error while proposing cluster id")
 					return
 				}
 				glog.Errorf("While proposing CID: %v. Retrying...", err)
@@ -507,6 +522,24 @@ func (n *node) initAndStartNode() error {
 	return nil
 }
 
+// periodically checks the validity of the enterprise license and updates the membership state.
+func (n *node) updateEnterpriseStatePeriodically(closer *y.Closer) {
+	defer closer.Done()
+
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
+
+	n.server.updateEnterpriseState()
+	for {
+		select {
+		case <-ticker.C:
+			n.server.updateEnterpriseState()
+		case <-closer.HasBeenClosed():
+			return
+		}
+	}
+}
+
 func (n *node) updateZeroMembershipPeriodically(closer *y.Closer) {
 	defer closer.Done()
 	ticker := time.NewTicker(10 * time.Second)
@@ -604,14 +637,15 @@ func (n *node) Run() {
 	// snapshot can cause select loop to block while deleting entries, so run
 	// it in goroutine
 	readStateCh := make(chan raft.ReadState, 100)
-	closer := y.NewCloser(4)
+	closer := y.NewCloser(5)
 	defer func() {
 		closer.SignalAndWait()
 		n.closer.Done()
 		glog.Infof("Zero Node.Run finished.")
 	}()
 
 	go n.snapshotPeriodically(closer)
+	go n.updateEnterpriseStatePeriodically(closer)
 	go n.updateZeroMembershipPeriodically(closer)
 	go n.checkQuorum(closer)
 	go n.RunReadIndexLoop(closer, readStateCh)

dgraph/cmd/zero/run.go

@@ -95,6 +95,8 @@ instances to achieve high-availability.
 	// about the status of supporting annotation logs through the datadog exporter
 	flag.String("datadog.collector", "", "Send opencensus traces to Datadog. As of now, the trace"+
 		" exporter does not support annotation logs and would discard them.")
+	// FIXME - Only for testing, remove before shipping.
+	flag.String("public_key", "", "Path to public key.")
 }
 
 func setupListener(addr string, port int, kind string) (listener net.Listener, err error) {
@@ -219,6 +221,7 @@ func run() {
 	http.HandleFunc("/removeNode", st.removeNode)
 	http.HandleFunc("/moveTablet", st.moveTablet)
 	http.HandleFunc("/assign", st.assign)
+	http.HandleFunc("/enterpriseLicense", st.applyEnterpriseLicense)
 	zpages.Handle(http.DefaultServeMux, "/z")
 
 	// This must be here. It does not work if placed before Grpc init.