From 10db4c7c8f1c3c68d12ebf6b77dc01cf7e657e48 Mon Sep 17 00:00:00 2001
From: Armon Dadgar <armon.dadgar@gmail.com>
Date: Tue, 12 Aug 2014 10:54:56 -0700
Subject: [PATCH] consul: Resolve parent ACLs

---
 consul/acl.go             | 21 ++++++++-------
 consul/acl_endpoint.go    |  6 ++---
 consul/acl_test.go        | 57 +++++++++++++++++++++++++++++++++++++++
 consul/structs/structs.go |  2 +-
 4 files changed, 72 insertions(+), 14 deletions(-)

diff --git a/consul/acl.go b/consul/acl.go
index f9609611b8..727ee0c5b0 100644
--- a/consul/acl.go
+++ b/consul/acl.go
@@ -100,7 +100,7 @@ func (s *Server) lookupACL(id, authDC string) (acl.ACL, error) {
 
 	// Handle the happy path
 	if err == nil {
-		return s.useACLPolicy(id, cached, &out)
+		return s.useACLPolicy(id, authDC, cached, &out)
 	}
 
 	// Check for not-found
@@ -125,7 +125,7 @@ func (s *Server) lookupACL(id, authDC string) (acl.ACL, error) {
 }
 
 // useACLPolicy handles an ACLPolicy response
-func (s *Server) useACLPolicy(id string, cached *aclCacheEntry, p *structs.ACLPolicy) (acl.ACL, error) {
+func (s *Server) useACLPolicy(id, authDC string, cached *aclCacheEntry, p *structs.ACLPolicy) (acl.ACL, error) {
 	// Check if we can used the cached policy
 	if cached != nil && cached.ETag == p.ETag {
 		if p.TTL > 0 {
@@ -140,17 +140,18 @@ func (s *Server) useACLPolicy(id string, cached *aclCacheEntry, p *structs.ACLPo
 	if ok {
 		compiled = raw.(acl.ACL)
 	} else {
-		// Determine the root policy
-		var root acl.ACL
-		switch p.Root {
-		case "allow":
-			root = acl.AllowAll()
-		default:
-			root = acl.DenyAll()
+		// Resolve the parent policy
+		parent := acl.RootACL(p.Parent)
+		if parent == nil {
+			var err error
+			parent, err = s.lookupACL(p.Parent, authDC)
+			if err != nil {
+				return nil, err
+			}
 		}
 
 		// Compile the ACL
-		acl, err := acl.New(root, p.Policy)
+		acl, err := acl.New(parent, p.Policy)
 		if err != nil {
 			return nil, err
 		}
diff --git a/consul/acl_endpoint.go b/consul/acl_endpoint.go
index 8761426331..67fec26c37 100644
--- a/consul/acl_endpoint.go
+++ b/consul/acl_endpoint.go
@@ -94,23 +94,23 @@ func (a *ACL) GetPolicy(args *structs.ACLPolicyRequest, reply *structs.ACLPolicy
 	}
 
 	// Get the policy via the cache
-	policy, err := a.srv.aclAuthCache.GetACLPolicy(args.ACL)
+	parent, policy, err := a.srv.aclAuthCache.GetACLPolicy(args.ACL)
 	if err != nil {
 		return err
 	}
 
 	// Generate an ETag
 	conf := a.srv.config
-	etag := fmt.Sprintf("%s:%s", conf.ACLDefaultPolicy, policy.ID)
+	etag := fmt.Sprintf("%s:%s", parent, policy.ID)
 
 	// Setup the response
 	reply.ETag = etag
-	reply.Root = conf.ACLDefaultPolicy
 	reply.TTL = conf.ACLTTL
 	a.srv.setQueryMeta(&reply.QueryMeta)
 
 	// Only send the policy on an Etag mis-match
 	if args.ETag != etag {
+		reply.Parent = parent
 		reply.Policy = policy
 	}
 	return nil
diff --git a/consul/acl_test.go b/consul/acl_test.go
index 85bcb907c7..7129a7e212 100644
--- a/consul/acl_test.go
+++ b/consul/acl_test.go
@@ -295,6 +295,63 @@ func TestACL_NonAuthority_Found(t *testing.T) {
 	}
 }
 
+func TestACL_NonAuthority_Management(t *testing.T) {
+	dir1, s1 := testServerWithConfig(t, func(c *Config) {
+		c.ACLDatacenter = "dc1" // Enable ACLs!
+		c.ACLMasterToken = "foobar"
+		c.ACLDefaultPolicy = "deny"
+	})
+	defer os.RemoveAll(dir1)
+	defer s1.Shutdown()
+	client := rpcClient(t, s1)
+	defer client.Close()
+
+	dir2, s2 := testServerWithConfig(t, func(c *Config) {
+		c.ACLDatacenter = "dc1" // Enable ACLs!
+		c.ACLDefaultPolicy = "deny"
+		c.Bootstrap = false // Disable bootstrap
+	})
+	defer os.RemoveAll(dir2)
+	defer s2.Shutdown()
+
+	// Try to join
+	addr := fmt.Sprintf("127.0.0.1:%d",
+		s1.config.SerfLANConfig.MemberlistConfig.BindPort)
+	if _, err := s2.JoinLAN([]string{addr}); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	testutil.WaitForResult(func() (bool, error) {
+		p1, _ := s1.raftPeers.Peers()
+		return len(p1) == 2, errors.New(fmt.Sprintf("%v", p1))
+	}, func(err error) {
+		t.Fatalf("should have 2 peers: %v", err)
+	})
+	testutil.WaitForLeader(t, client.Call, "dc1")
+
+	// find the non-authoritative server
+	var nonAuth *Server
+	if !s1.IsLeader() {
+		nonAuth = s1
+	} else {
+		nonAuth = s2
+	}
+
+	// Resolve the token
+	acl, err := nonAuth.resolveToken("foobar")
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if acl == nil {
+		t.Fatalf("missing acl")
+	}
+
+	// Check the policy, should allow all
+	if !acl.KeyRead("foo/test") {
+		t.Fatalf("unexpected failed read")
+	}
+}
+
 func TestACL_DownPolicy_Deny(t *testing.T) {
 	dir1, s1 := testServerWithConfig(t, func(c *Config) {
 		c.ACLDatacenter = "dc1"
diff --git a/consul/structs/structs.go b/consul/structs/structs.go
index 3263a6ce28..95f273f4fc 100644
--- a/consul/structs/structs.go
+++ b/consul/structs/structs.go
@@ -487,7 +487,7 @@ type IndexedACLs struct {
 
 type ACLPolicy struct {
 	ETag   string
-	Root   string
+	Parent string
 	Policy *acl.Policy
 	TTL    time.Duration
 	QueryMeta