fix(ACL): The Acl cache should be updated on restart and restore

dgraph/cmd/alpha/run.go

@@ -818,8 +818,8 @@ func run() {
 
 		// initialization of the admin account can only be done after raft nodes are running
 		// and health check passes
-		edgraph.ResetAcl(updaters)
-		edgraph.RefreshAcls(updaters)
+		edgraph.InitializeAcl(updaters)
+		edgraph.SubscribeForAclUpdates(updaters)
 	}()
 
 	// Graphql subscribes to alpha to get schema updates. We need to close that before we

edgraph/access.go

@@ -50,13 +50,18 @@ func upsertGuardianAndGroot(closer *z.Closer, ns uint64) {
 	// do nothing
 }
 
-// ResetAcls is an empty method since ACL is only supported in the enterprise version.
-func RefreshAcls(closer *z.Closer) {
+// SubscribeForAclUpdates is an empty method since ACL is only supported in the enterprise version.
+func SubscribeForAclUpdates(closer *z.Closer) {
 	// do nothing
 	<-closer.HasBeenClosed()
 	closer.Done()
 }
 
+// RefreshACLs is an empty method since ACL is only supported in the enterprise version.
+func RefreshACLs(ctx context.Context) {
+	return
+}
+
 func authorizeAlter(ctx context.Context, op *api.Operation) error {
 	return nil
 }

edgraph/access_ee.go

@@ -316,8 +316,44 @@ func authorizeUser(ctx context.Context, userid string, password string) (
 	return user, nil
 }
 
-// RefreshAcls queries for the ACL triples and refreshes the ACLs accordingly.
-func RefreshAcls(closer *z.Closer) {
+func refreshAclCache(ctx context.Context, ns, refreshTs uint64) error {
+	glog.V(2).Infof("Refreshing ACLs")
+	req := &Request{
+		req: &api.Request{
+			Query:    queryAcls,
+			ReadOnly: true,
+			StartTs:  refreshTs,
+		},
+		doAuth: NoAuthorize,
+	}
+
+	ctx = x.AttachNamespace(ctx, ns)
+	queryResp, err := (&Server{}).doQuery(ctx, req)
+	if err != nil {
+		return errors.Errorf("unable to retrieve acls: %v", err)
+	}
+	groups, err := acl.UnmarshalGroups(queryResp.GetJson(), "allAcls")
+	if err != nil {
+		return err
+	}
+
+	worker.AclCachePtr.Update(ns, groups)
+	glog.V(2).Infof("Updated the ACL cache for namespace: %#x", ns)
+	return nil
+
+}
+
+func RefreshACLs(ctx context.Context) {
+	for ns := range schema.State().Namespaces() {
+		if err := refreshAclCache(ctx, ns, 0); err != nil {
+			glog.Errorf("Error while retrieving acls for namespace %#x: %v", ns, err)
+		}
+	}
+	worker.AclCachePtr.Set()
+}
+
+// SubscribeForAclUpdates subscribes for ACL predicates and updates the acl cache.
+func SubscribeForAclUpdates(closer *z.Closer) {
 	defer func() {
 		glog.Infoln("RefreshAcls closed")
 		closer.Done()
@@ -327,38 +363,20 @@ func RefreshAcls(closer *z.Closer) {
 		return
 	}
 
-	// retrieve the full data set of ACLs from the corresponding alpha server, and update the
-	// aclCachePtr
 	var maxRefreshTs uint64
 	retrieveAcls := func(ns uint64, refreshTs uint64) error {
 		if refreshTs <= maxRefreshTs {
 			return nil
 		}
 		maxRefreshTs = refreshTs
-
-		glog.V(3).Infof("Refreshing ACLs")
-		req := &Request{
-			req: &api.Request{
-				Query:    queryAcls,
-				ReadOnly: true,
-				StartTs:  refreshTs,
-			},
-			doAuth: NoAuthorize,
-		}
-
 		ctx := x.AttachNamespace(closer.Ctx(), ns)
-		queryResp, err := (&Server{}).doQuery(ctx, req)
-		if err != nil {
-			return errors.Errorf("unable to retrieve acls: %v", err)
-		}
-		groups, err := acl.UnmarshalGroups(queryResp.GetJson(), "allAcls")
-		if err != nil {
-			return err
+
+		if !worker.AclCachePtr.Loaded() {
+			updaters := z.NewCloser(1)
+			RefreshACLs(updaters.Ctx())
 		}
 
-		aclCachePtr.update(ns, groups)
-		glog.V(3).Infof("Updated the ACL cache")
-		return nil
+		return refreshAclCache(ctx, ns, refreshTs)
 	}
 
 	closer.AddRunning(1)
@@ -406,8 +424,8 @@ var aclPrefixes = [][]byte{
 	x.PredicatePrefix(x.GalaxyAttr("dgraph.xid")),
 }
 
-// clears the aclCachePtr and upserts the Groot account.
-func ResetAcl(closer *z.Closer) {
+// upserts the Groot account.
+func InitializeAcl(closer *z.Closer) {
 	defer func() {
 		glog.Infof("ResetAcl closed")
 		closer.Done()
@@ -613,13 +631,17 @@ type authPredResult struct {
 func authorizePreds(ctx context.Context, userData *userData, preds []string,
 	aclOp *acl.Operation) *authPredResult {
 
+	if !worker.AclCachePtr.Loaded() {
+		RefreshACLs(ctx)
+	}
+
 	userId := userData.userId
 	groupIds := userData.groupIds
 	ns := userData.namespace
 	blockedPreds := make(map[string]struct{})
 	for _, pred := range preds {
 		nsPred := x.NamespaceAttr(ns, pred)
-		if err := aclCachePtr.authorizePredicate(groupIds, nsPred, aclOp); err != nil {
+		if err := worker.AclCachePtr.AuthorizePredicate(groupIds, nsPred, aclOp); err != nil {
 			logAccess(&accessEntry{
 				userId:    userId,
 				groups:    groupIds,
@@ -631,21 +653,21 @@ func authorizePreds(ctx context.Context, userData *userData, preds []string,
 			blockedPreds[pred] = struct{}{}
 		}
 	}
-	aclCachePtr.RLock()
-	allowedPreds := make([]string, len(aclCachePtr.userPredPerms[userId]))
+	worker.AclCachePtr.RLock()
 	// User can have multiple permission for same predicate, add predicate
+	allowedPreds := make([]string, len(worker.AclCachePtr.GetUserPredPerms(userId)))
 	// only if the acl.Op is covered in the set of permissions for the user
-	for predicate, perm := range aclCachePtr.userPredPerms[userId] {
+	for predicate, perm := range worker.AclCachePtr.GetUserPredPerms(userId) {
 		if (perm & aclOp.Code) > 0 {
 			allowedPreds = append(allowedPreds, predicate)
 		}
 	}
-	aclCachePtr.RUnlock()
+	worker.AclCachePtr.RUnlock()
 	return &authPredResult{allowed: allowedPreds, blocked: blockedPreds}
 }
 
 // authorizeAlter parses the Schema in the operation and authorizes the operation
-// using the aclCachePtr. It will return error if any one of the predicates specified in alter
+// using the worker.AclCachePtr. It will return error if any one of the predicates specified in alter
 // are not authorized.
 func authorizeAlter(ctx context.Context, op *api.Operation) error {
 	if len(worker.Config.HmacSecret) == 0 {
@@ -768,7 +790,7 @@ func isAclPredMutation(nquads []*api.NQuad) bool {
 	return false
 }
 
-// authorizeMutation authorizes the mutation using the aclCachePtr. It will return permission
+// authorizeMutation authorizes the mutation using the worker.AclCachePtr. It will return permission
 // denied error if any one of the predicates in mutation(set or delete) is unauthorized.
 // At this stage, namespace is not attached in the predicates.
 func authorizeMutation(ctx context.Context, gmu *gql.Mutation) error {
@@ -928,7 +950,7 @@ func shouldAllowAcls(ns uint64) bool {
 	return !x.Config.SharedInstance || ns == x.GalaxyNamespace
 }
 
-// authorizeQuery authorizes the query using the aclCachePtr. It will silently drop all
+// authorizeQuery authorizes the query using the worker.AclCachePtr. It will silently drop all
 // unauthorized predicates from query.
 // At this stage, namespace is not attached in the predicates.
 func authorizeQuery(ctx context.Context, parsedReq *gql.Result, graphql bool) error {

edgraph/server.go

@@ -409,7 +409,7 @@ func (s *Server) Alter(ctx context.Context, op *api.Operation) (*api.Payload, er
 		// reset their in-memory GraphQL schema
 		_, err = UpdateGQLSchema(ctx, "", "")
 		// recreate the admin account after a drop all operation
-		ResetAcl(nil)
+		InitializeAcl(nil)
 		return empty, err
 	}
 

graphql/admin/restore.go

@@ -84,7 +84,7 @@ func resolveRestore(ctx context.Context, m schema.Mutation) (*resolve.Resolved,
 
 	go func() {
 		wg.Wait()
-		edgraph.ResetAcl(nil)
+		edgraph.InitializeAcl(nil)
 	}()
 
 	return resolve.DataResult(

edgraph/acl_cache.go → worker/acl_cache.go

@@ -1,16 +1,4 @@
-// +build !oss
-
-/*
- * Copyright 2018 Dgraph Labs, Inc. All rights reserved.
- *
- * Licensed under the Dgraph Community License (the "License"); you
- * may not use this file except in compliance with the License. You
- * may obtain a copy of the License at
- *
- *     https://github.com/dgraph-io/dgraph/blob/master/licenses/DCL.txt
- */
-
-package edgraph
+package worker
 
 import (
 	"sync"
@@ -21,22 +9,44 @@ import (
 )
 
 // aclCache is the cache mapping group names to the corresponding group acls
-type aclCache struct {
+type AclCache struct {
 	sync.RWMutex
+	loaded        bool
 	predPerms     map[string]map[string]int32
 	userPredPerms map[string]map[string]int32
 }
 
-var aclCachePtr = &aclCache{
+func (cache *AclCache) reset() {
+	cache.loaded = false
+}
+
+func ResetAclCache() {
+	AclCachePtr.reset()
+}
+
+func (cache *AclCache) Loaded() bool {
+	return cache.loaded
+}
+
+func (cache *AclCache) Set() {
+	cache.loaded = true
+}
+
+var AclCachePtr = &AclCache{
+	loaded:        false,
 	predPerms:     make(map[string]map[string]int32),
 	userPredPerms: make(map[string]map[string]int32),
 }
 
-func (cache *aclCache) update(ns uint64, groups []acl.Group) {
+func (cache *AclCache) GetUserPredPerms(userId string) map[string]int32 {
+	return cache.userPredPerms[userId]
+}
+
+func (cache *AclCache) Update(ns uint64, groups []acl.Group) {
 	// In dgraph, acl rules are divided by groups, e.g.
 	// the dev group has the following blob representing its ACL rules
 	// [friend, 4], [name, 7] where friend and name are predicates,
-	// However in the aclCachePtr in memory, we need to change the structure and store
+	// However in the AclCachePtr in memory, we need to change the structure and store
 	// the information in two formats for efficient look-ups.
 	//
 	// First in which ACL rules are divided by predicates, e.g.
@@ -101,21 +111,21 @@ func (cache *aclCache) update(ns uint64, groups []acl.Group) {
 		}
 	}
 
-	aclCachePtr.Lock()
-	defer aclCachePtr.Unlock()
-	aclCachePtr.predPerms = predPerms
-	aclCachePtr.userPredPerms = userPredPerms
+	AclCachePtr.Lock()
+	defer AclCachePtr.Unlock()
+	AclCachePtr.predPerms = predPerms
+	AclCachePtr.userPredPerms = userPredPerms
 }
 
-func (cache *aclCache) authorizePredicate(groups []string, predicate string,
+func (cache *AclCache) AuthorizePredicate(groups []string, predicate string,
 	operation *acl.Operation) error {
 	if x.IsAclPredicate(x.ParseAttr(predicate)) {
 		return errors.Errorf("only groot is allowed to access the ACL predicate: %s", predicate)
 	}
 
-	aclCachePtr.RLock()
-	predPerms := aclCachePtr.predPerms
-	aclCachePtr.RUnlock()
+	AclCachePtr.RLock()
+	predPerms := AclCachePtr.predPerms
+	AclCachePtr.RUnlock()
 
 	if groupPerms, found := predPerms[predicate]; found {
 		if hasRequiredAccess(groupPerms, groups, operation) {

edgraph/acl_cache_test.go → worker/acl_cache_test.go

@@ -10,7 +10,7 @@
  *     https://github.com/dgraph-io/dgraph/blob/master/licenses/DCL.txt
  */
 
-package edgraph
+package worker
 
 import (
 	"testing"
@@ -21,14 +21,14 @@ import (
 )
 
 func TestAclCache(t *testing.T) {
-	aclCachePtr = &aclCache{
+	AclCachePtr = &AclCache{
 		predPerms: make(map[string]map[string]int32),
 	}
 
 	var emptyGroups []string
 	group := "dev"
 	predicate := x.GalaxyAttr("friend")
-	require.Error(t, aclCachePtr.authorizePredicate(emptyGroups, predicate, acl.Read),
+	require.Error(t, AclCachePtr.AuthorizePredicate(emptyGroups, predicate, acl.Read),
 		"the anonymous user should not have access when the acl cache is empty")
 
 	acls := []acl.Acl{
@@ -44,16 +44,16 @@ func TestAclCache(t *testing.T) {
 			Rules:   acls,
 		},
 	}
-	aclCachePtr.update(x.GalaxyNamespace, groups)
+	AclCachePtr.Update(x.GalaxyNamespace, groups)
 	// after a rule is defined, the anonymous user should no longer have access
-	require.Error(t, aclCachePtr.authorizePredicate(emptyGroups, predicate, acl.Read),
+	require.Error(t, AclCachePtr.AuthorizePredicate(emptyGroups, predicate, acl.Read),
 		"the anonymous user should not have access when the predicate has acl defined")
-	require.NoError(t, aclCachePtr.authorizePredicate([]string{group}, predicate, acl.Read),
+	require.NoError(t, AclCachePtr.AuthorizePredicate([]string{group}, predicate, acl.Read),
 		"the user with group authorized should have access")
 
 	// update the cache with empty acl list in order to clear the cache
-	aclCachePtr.update(x.GalaxyNamespace, []acl.Group{})
+	AclCachePtr.Update(x.GalaxyNamespace, []acl.Group{})
 	// the anonymous user should have access again
-	require.Error(t, aclCachePtr.authorizePredicate(emptyGroups, predicate, acl.Read),
+	require.Error(t, AclCachePtr.AuthorizePredicate(emptyGroups, predicate, acl.Read),
 		"the anonymous user should not have access when the acl cache is empty")
 }

worker/online_restore.go

@@ -339,6 +339,7 @@ func handleRestoreProposal(ctx context.Context, req *pb.RestoreRequest) error {
 	// reset gql schema
 	glog.Info("reseting local gql schema store")
 	ResetGQLSchemaStore()
+	ResetAclCache()
 
 	// Propose a snapshot immediately after all the work is done to prevent the restore
 	// from being replayed.