func TestACL_filterNodes(t *testing.T) {
	// Create a nodes list.
	nodes := structs.Nodes{
		&structs.Node{
			Node: "foo",
		},
		&structs.Node{
			Node: "bar",
		},
	}

	// Try permissive filtering.
	filt := newAclFilter(acl.AllowAll(), nil, true)
	filt.filterNodes(&nodes)
	if len(nodes) != 2 {
		t.Fatalf("bad: %#v", nodes)
	}

	// Try restrictive filtering but with version 8 enforcement turned off.
	filt = newAclFilter(acl.DenyAll(), nil, false)
	filt.filterNodes(&nodes)
	if len(nodes) != 2 {
		t.Fatalf("bad: %#v", nodes)
	}

	// Try restrictive filtering with version 8 enforcement turned on.
	filt = newAclFilter(acl.DenyAll(), nil, true)
	filt.filterNodes(&nodes)
	if len(nodes) != 0 {
		t.Fatalf("bad: %#v", nodes)
	}
}

func TestACL_filterNodeServices(t *testing.T) {
	// Create some node services
	services := structs.NodeServices{
		Node: &structs.Node{
			Node: "node1",
		},
		Services: map[string]*structs.NodeService{
			"foo": &structs.NodeService{
				ID:      "foo",
				Service: "foo",
			},
		},
	}

	// Try permissive filtering
	filt := newAclFilter(acl.AllowAll(), nil)
	filt.filterNodeServices(&services)
	if len(services.Services) != 1 {
		t.Fatalf("bad: %#v", services.Services)
	}

	// Try restrictive filtering
	filt = newAclFilter(acl.DenyAll(), nil)
	filt.filterNodeServices(&services)
	if len(services.Services) != 0 {
		t.Fatalf("bad: %#v", services.Services)
	}
}

func TestACL_filterCoordinates(t *testing.T) {
	// Create some coordinates.
	coords := structs.Coordinates{
		&structs.Coordinate{
			Node:  "node1",
			Coord: generateRandomCoordinate(),
		},
		&structs.Coordinate{
			Node:  "node2",
			Coord: generateRandomCoordinate(),
		},
	}

	// Try permissive filtering.
	filt := newAclFilter(acl.AllowAll(), nil, false)
	filt.filterCoordinates(&coords)
	if len(coords) != 2 {
		t.Fatalf("bad: %#v", coords)
	}

	// Try restrictive filtering without version 8 ACL enforcement.
	filt = newAclFilter(acl.DenyAll(), nil, false)
	filt.filterCoordinates(&coords)
	if len(coords) != 2 {
		t.Fatalf("bad: %#v", coords)
	}

	// Try restrictive filtering with version 8 ACL enforcement.
	filt = newAclFilter(acl.DenyAll(), nil, true)
	filt.filterCoordinates(&coords)
	if len(coords) != 0 {
		t.Fatalf("bad: %#v", coords)
	}
}

// lookupACL is used when we are non-authoritative, and need
// to resolve an ACL
func (s *Server) lookupACL(id, authDC string) (acl.ACL, error) {
	// Check the cache for the ACL
	var cached *aclCacheEntry
	raw, ok := s.aclCache.Get(id)
	if ok {
		cached = raw.(*aclCacheEntry)
	}

	// Check for live cache
	if cached != nil && time.Now().Before(cached.Expires) {
		metrics.IncrCounter([]string{"consul", "acl", "cache_hit"}, 1)
		return cached.ACL, nil
	} else {
		metrics.IncrCounter([]string{"consul", "acl", "cache_miss"}, 1)
	}

	// Attempt to refresh the policy
	args := structs.ACLPolicyRequest{
		Datacenter: authDC,
		ACL:        id,
	}
	if cached != nil {
		args.ETag = cached.ETag
	}
	var out structs.ACLPolicy
	err := s.RPC("ACL.GetPolicy", &args, &out)

	// Handle the happy path
	if err == nil {
		return s.useACLPolicy(id, authDC, cached, &out)
	}

	// Check for not-found
	if strings.Contains(err.Error(), aclNotFound) {
		return nil, errors.New(aclNotFound)
	} else {
		s.logger.Printf("[ERR] consul.acl: Failed to get policy for '%s': %v", id, err)
	}

	// Unable to refresh, apply the down policy
	switch s.config.ACLDownPolicy {
	case "allow":
		return acl.AllowAll(), nil
	case "extend-cache":
		if cached != nil {
			return cached.ACL, nil
		}
		fallthrough
	default:
		return acl.DenyAll(), nil
	}
}

func TestACL_filterNodeDump(t *testing.T) {
	// Create a node dump
	dump := structs.NodeDump{
		&structs.NodeInfo{
			Node: "node1",
			Services: []*structs.NodeService{
				&structs.NodeService{
					ID:      "foo",
					Service: "foo",
				},
			},
			Checks: []*structs.HealthCheck{
				&structs.HealthCheck{
					Node:        "node1",
					CheckID:     "check1",
					ServiceName: "foo",
				},
			},
		},
	}

	// Try permissive filtering
	filt := newAclFilter(acl.AllowAll(), nil)
	filt.filterNodeDump(&dump)
	if len(dump) != 1 {
		t.Fatalf("bad: %#v", dump)
	}
	if len(dump[0].Services) != 1 {
		t.Fatalf("bad: %#v", dump[0].Services)
	}
	if len(dump[0].Checks) != 1 {
		t.Fatalf("bad: %#v", dump[0].Checks)
	}

	// Try restrictive filtering
	filt = newAclFilter(acl.DenyAll(), nil)
	filt.filterNodeDump(&dump)
	if len(dump) != 1 {
		t.Fatalf("bad: %#v", dump)
	}
	if len(dump[0].Services) != 0 {
		t.Fatalf("bad: %#v", dump[0].Services)
	}
	if len(dump[0].Checks) != 0 {
		t.Fatalf("bad: %#v", dump[0].Checks)
	}
}

// newACLManager returns an ACL manager based on the given config.
func newACLManager(config *Config) (*aclManager, error) {
	// Set up the cache from ID to ACL (we don't cache policies like the
	// servers; only one level).
	acls, err := lru.New2Q(aclCacheSize)
	if err != nil {
		return nil, err
	}

	// If an agent master token is configured, build a policy and ACL for
	// it, otherwise leave it nil.
	var master acl.ACL
	if len(config.ACLAgentMasterToken) > 0 {
		policy := &acl.Policy{
			Agents: []*acl.AgentPolicy{
				&acl.AgentPolicy{
					Node:   config.NodeName,
					Policy: acl.PolicyWrite,
				},
			},
		}
		acl, err := acl.New(acl.DenyAll(), policy)
		if err != nil {
			return nil, err
		}
		master = acl
	}

	var down acl.ACL
	switch config.ACLDownPolicy {
	case "allow":
		down = acl.AllowAll()
	case "deny":
		down = acl.DenyAll()
	case "extend-cache":
		// Leave the down policy as nil to signal this.
	default:
		return nil, fmt.Errorf("invalid ACL down policy %q", config.ACLDownPolicy)
	}

	// Give back a manager.
	return &aclManager{
		acls:   acls,
		master: master,
		down:   down,
	}, nil
}

func TestACL_filterServices(t *testing.T) {
	// Create some services
	services := structs.Services{
		"service1": []string{},
		"service2": []string{},
	}

	// Try permissive filtering
	filt := newAclFilter(acl.AllowAll(), nil)
	filt.filterServices(services)
	if len(services) != 2 {
		t.Fatalf("bad: %#v", services)
	}

	// Try restrictive filtering
	filt = newAclFilter(acl.DenyAll(), nil)
	filt.filterServices(services)
	if len(services) != 0 {
		t.Fatalf("bad: %#v", services)
	}
}

func TestACL_filterCheckServiceNodes(t *testing.T) {
	// Create some nodes
	nodes := structs.CheckServiceNodes{
		structs.CheckServiceNode{
			Node: &structs.Node{
				Node: "node1",
			},
			Service: &structs.NodeService{
				ID:      "foo",
				Service: "foo",
			},
			Checks: structs.HealthChecks{
				&structs.HealthCheck{
					Node:        "node1",
					CheckID:     "check1",
					ServiceName: "foo",
				},
			},
		},
	}

	// Try permissive filtering
	filt := newAclFilter(acl.AllowAll(), nil)
	filt.filterCheckServiceNodes(&nodes)
	if len(nodes) != 1 {
		t.Fatalf("bad: %#v", nodes)
	}
	if len(nodes[0].Checks) != 1 {
		t.Fatalf("bad: %#v", nodes[0].Checks)
	}

	// Try restrictive filtering
	filt = newAclFilter(acl.DenyAll(), nil)
	filt.filterCheckServiceNodes(&nodes)
	if len(nodes) != 0 {
		t.Fatalf("bad: %#v", nodes)
	}
}

func TestACL_filterServiceNodes(t *testing.T) {
	// Create some service nodes
	nodes := structs.ServiceNodes{
		&structs.ServiceNode{
			Node:        "node1",
			ServiceName: "foo",
		},
	}

	// Try permissive filtering
	filt := newAclFilter(acl.AllowAll(), nil)
	filt.filterServiceNodes(&nodes)
	if len(nodes) != 1 {
		t.Fatalf("bad: %#v", nodes)
	}

	// Try restrictive filtering
	filt = newAclFilter(acl.DenyAll(), nil)
	filt.filterServiceNodes(&nodes)
	if len(nodes) != 0 {
		t.Fatalf("bad: %#v", nodes)
	}
}

func TestACL_redactPreparedQueryTokens(t *testing.T) {
	query := &structs.PreparedQuery{
		ID:    "f004177f-2c28-83b7-4229-eacc25fe55d1",
		Token: "root",
	}

	expected := &structs.PreparedQuery{
		ID:    "f004177f-2c28-83b7-4229-eacc25fe55d1",
		Token: "root",
	}

	// Try permissive filtering with a management token. This will allow the
	// embedded token to be seen.
	filt := newAclFilter(acl.ManageAll(), nil)
	filt.redactPreparedQueryTokens(&query)
	if !reflect.DeepEqual(query, expected) {
		t.Fatalf("bad: %#v", &query)
	}

	// Hang on to the entry with a token, which needs to survive the next
	// operation.
	original := query

	// Now try permissive filtering with a client token, which should cause
	// the embedded token to get redacted.
	filt = newAclFilter(acl.AllowAll(), nil)
	filt.redactPreparedQueryTokens(&query)
	expected.Token = redactedToken
	if !reflect.DeepEqual(query, expected) {
		t.Fatalf("bad: %#v", *query)
	}

	// Make sure that the original object didn't lose its token.
	if original.Token != "root" {
		t.Fatalf("bad token: %s", original.Token)
	}
}

func TestACL_filterHealthChecks(t *testing.T) {
	// Create some health checks
	hc := structs.HealthChecks{
		&structs.HealthCheck{
			Node:        "node1",
			CheckID:     "check1",
			ServiceName: "foo",
		},
	}

	// Try permissive filtering
	filt := newAclFilter(acl.AllowAll(), nil)
	filt.filterHealthChecks(&hc)
	if len(hc) != 1 {
		t.Fatalf("bad: %#v", hc)
	}

	// Try restrictive filtering
	filt = newAclFilter(acl.DenyAll(), nil)
	filt.filterHealthChecks(&hc)
	if len(hc) != 0 {
		t.Fatalf("bad: %#v", hc)
	}
}

func TestACL_DownPolicy_Allow(t *testing.T) {
	dir1, s1 := testServerWithConfig(t, func(c *Config) {
		c.ACLDatacenter = "dc1"
		c.ACLDownPolicy = "allow"
		c.ACLMasterToken = "root"
	})
	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.ACLDownPolicy = "allow"
		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, s1.RPC, "dc1")

	// Create a new token
	arg := structs.ACLRequest{
		Datacenter: "dc1",
		Op:         structs.ACLSet,
		ACL: structs.ACL{
			Name:  "User token",
			Type:  structs.ACLTypeClient,
			Rules: testACLPolicy,
		},
		WriteRequest: structs.WriteRequest{Token: "root"},
	}
	var id string
	if err := s1.RPC("ACL.Apply", &arg, &id); err != nil {
		t.Fatalf("err: %v", err)
	}

	// find the non-authoritative server
	var nonAuth *Server
	var auth *Server
	if !s1.IsLeader() {
		nonAuth = s1
		auth = s2
	} else {
		nonAuth = s2
		auth = s1
	}

	// Kill the authoritative server
	auth.Shutdown()

	// Token should resolve into a AllowAll
	aclR, err := nonAuth.resolveToken(id)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	if aclR != acl.AllowAll() {
		t.Fatalf("bad acl: %#v", aclR)
	}
}

func TestACL_filterPreparedQueries(t *testing.T) {
	queries := structs.PreparedQueries{
		&structs.PreparedQuery{
			ID: "f004177f-2c28-83b7-4229-eacc25fe55d1",
		},
		&structs.PreparedQuery{
			ID:   "f004177f-2c28-83b7-4229-eacc25fe55d2",
			Name: "query-with-no-token",
		},
		&structs.PreparedQuery{
			ID:    "f004177f-2c28-83b7-4229-eacc25fe55d3",
			Name:  "query-with-a-token",
			Token: "root",
		},
	}

	expected := structs.PreparedQueries{
		&structs.PreparedQuery{
			ID: "f004177f-2c28-83b7-4229-eacc25fe55d1",
		},
		&structs.PreparedQuery{
			ID:   "f004177f-2c28-83b7-4229-eacc25fe55d2",
			Name: "query-with-no-token",
		},
		&structs.PreparedQuery{
			ID:    "f004177f-2c28-83b7-4229-eacc25fe55d3",
			Name:  "query-with-a-token",
			Token: "root",
		},
	}

	// Try permissive filtering with a management token. This will allow the
	// embedded token to be seen.
	filt := newAclFilter(acl.ManageAll(), nil)
	filt.filterPreparedQueries(&queries)
	if !reflect.DeepEqual(queries, expected) {
		t.Fatalf("bad: %#v", queries)
	}

	// Hang on to the entry with a token, which needs to survive the next
	// operation.
	original := queries[2]

	// Now try permissive filtering with a client token, which should cause
	// the embedded token to get redacted, and the query with no name to get
	// filtered out.
	filt = newAclFilter(acl.AllowAll(), nil)
	filt.filterPreparedQueries(&queries)
	expected[2].Token = redactedToken
	expected = append(structs.PreparedQueries{}, expected[1], expected[2])
	if !reflect.DeepEqual(queries, expected) {
		t.Fatalf("bad: %#v", queries)
	}

	// Make sure that the original object didn't lose its token.
	if original.Token != "root" {
		t.Fatalf("bad token: %s", original.Token)
	}

	// Now try restrictive filtering.
	filt = newAclFilter(acl.DenyAll(), nil)
	filt.filterPreparedQueries(&queries)
	if len(queries) != 0 {
		t.Fatalf("bad: %#v", queries)
	}
}

func TestACL_filterServiceNodes(t *testing.T) {
	// Create some service nodes.
	fill := func() structs.ServiceNodes {
		return structs.ServiceNodes{
			&structs.ServiceNode{
				Node:        "node1",
				ServiceName: "foo",
			},
		}
	}

	// Try permissive filtering.
	{
		nodes := fill()
		filt := newAclFilter(acl.AllowAll(), nil, false)
		filt.filterServiceNodes(&nodes)
		if len(nodes) != 1 {
			t.Fatalf("bad: %#v", nodes)
		}
	}

	// Try restrictive filtering.
	{
		nodes := fill()
		filt := newAclFilter(acl.DenyAll(), nil, false)
		filt.filterServiceNodes(&nodes)
		if len(nodes) != 0 {
			t.Fatalf("bad: %#v", nodes)
		}
	}

	// Allowed to see the service but not the node.
	policy, err := acl.Parse(`
service "foo" {
  policy = "read"
}
`)
	if err != nil {
		t.Fatalf("err %v", err)
	}
	perms, err := acl.New(acl.DenyAll(), policy)
	if err != nil {
		t.Fatalf("err: %v", err)
	}

	// This will work because version 8 ACLs aren't being enforced.
	{
		nodes := fill()
		filt := newAclFilter(perms, nil, false)
		filt.filterServiceNodes(&nodes)
		if len(nodes) != 1 {
			t.Fatalf("bad: %#v", nodes)
		}
	}

	// But with version 8 the node will block it.
	{
		nodes := fill()
		filt := newAclFilter(perms, nil, true)
		filt.filterServiceNodes(&nodes)
		if len(nodes) != 0 {
			t.Fatalf("bad: %#v", nodes)
		}
	}

	// Chain on access to the node.
	policy, err = acl.Parse(`
node "node1" {
  policy = "read"
}
`)
	if err != nil {
		t.Fatalf("err %v", err)
	}
	perms, err = acl.New(perms, policy)
	if err != nil {
		t.Fatalf("err: %v", err)
	}

	// Now it should go through.
	{
		nodes := fill()
		filt := newAclFilter(perms, nil, true)
		filt.filterServiceNodes(&nodes)
		if len(nodes) != 1 {
			t.Fatalf("bad: %#v", nodes)
		}
	}
}

func TestACL_filterNodeDump(t *testing.T) {
	// Create a node dump.
	fill := func() structs.NodeDump {
		return structs.NodeDump{
			&structs.NodeInfo{
				Node: "node1",
				Services: []*structs.NodeService{
					&structs.NodeService{
						ID:      "foo",
						Service: "foo",
					},
				},
				Checks: []*structs.HealthCheck{
					&structs.HealthCheck{
						Node:        "node1",
						CheckID:     "check1",
						ServiceName: "foo",
					},
				},
			},
		}
	}

	// Try permissive filtering.
	{
		dump := fill()
		filt := newAclFilter(acl.AllowAll(), nil, false)
		filt.filterNodeDump(&dump)
		if len(dump) != 1 {
			t.Fatalf("bad: %#v", dump)
		}
		if len(dump[0].Services) != 1 {
			t.Fatalf("bad: %#v", dump[0].Services)
		}
		if len(dump[0].Checks) != 1 {
			t.Fatalf("bad: %#v", dump[0].Checks)
		}
	}

	// Try restrictive filtering.
	{
		dump := fill()
		filt := newAclFilter(acl.DenyAll(), nil, false)
		filt.filterNodeDump(&dump)
		if len(dump) != 1 {
			t.Fatalf("bad: %#v", dump)
		}
		if len(dump[0].Services) != 0 {
			t.Fatalf("bad: %#v", dump[0].Services)
		}
		if len(dump[0].Checks) != 0 {
			t.Fatalf("bad: %#v", dump[0].Checks)
		}
	}

	// Allowed to see the service but not the node.
	policy, err := acl.Parse(`
service "foo" {
  policy = "read"
}
`)
	if err != nil {
		t.Fatalf("err %v", err)
	}
	perms, err := acl.New(acl.DenyAll(), policy)
	if err != nil {
		t.Fatalf("err: %v", err)
	}

	// This will work because version 8 ACLs aren't being enforced.
	{
		dump := fill()
		filt := newAclFilter(perms, nil, false)
		filt.filterNodeDump(&dump)
		if len(dump) != 1 {
			t.Fatalf("bad: %#v", dump)
		}
		if len(dump[0].Services) != 1 {
			t.Fatalf("bad: %#v", dump[0].Services)
		}
		if len(dump[0].Checks) != 1 {
			t.Fatalf("bad: %#v", dump[0].Checks)
		}
	}

	// But with version 8 the node will block it.
	{
		dump := fill()
		filt := newAclFilter(perms, nil, true)
		filt.filterNodeDump(&dump)
		if len(dump) != 0 {
			t.Fatalf("bad: %#v", dump)
		}
	}

	// Chain on access to the node.
	policy, err = acl.Parse(`
node "node1" {
  policy = "read"
}
//.........这里部分代码省略.........

// lookupACL is used when we are non-authoritative, and need to resolve an ACL.
func (c *aclCache) lookupACL(id, authDC string) (acl.ACL, error) {
	// Check the cache for the ACL.
	var cached *aclCacheEntry
	raw, ok := c.acls.Get(id)
	if ok {
		cached = raw.(*aclCacheEntry)
	}

	// Check for live cache.
	if cached != nil && time.Now().Before(cached.Expires) {
		metrics.IncrCounter([]string{"consul", "acl", "cache_hit"}, 1)
		return cached.ACL, nil
	} else {
		metrics.IncrCounter([]string{"consul", "acl", "cache_miss"}, 1)
	}

	// Attempt to refresh the policy from the ACL datacenter via an RPC.
	args := structs.ACLPolicyRequest{
		Datacenter: authDC,
		ACL:        id,
	}
	if cached != nil {
		args.ETag = cached.ETag
	}
	var reply structs.ACLPolicy
	err := c.rpc("ACL.GetPolicy", &args, &reply)
	if err == nil {
		return c.useACLPolicy(id, authDC, cached, &reply)
	}

	// Check for not-found, which will cause us to bail immediately. For any
	// other error we report it in the logs but can continue.
	if strings.Contains(err.Error(), aclNotFound) {
		return nil, errors.New(aclNotFound)
	} else {
		c.logger.Printf("[ERR] consul.acl: Failed to get policy from ACL datacenter: %v", err)
	}

	// TODO (slackpad) - We could do a similar thing *within* the ACL
	// datacenter if the leader isn't available. We have a local state
	// store of the ACLs, so by populating the local member in this cache,
	// it would fall back to the state store if there was a leader loss and
	// the extend-cache policy was true. This feels subtle to explain and
	// configure, and leader blips should be paved over by cache already, so
	// we won't do this for now but should consider for the future. This is
	// a lot different than the replication story where you might be cut off
	// from the ACL datacenter for an extended period of time and need to
	// carry on operating with the full set of ACLs as they were known
	// before the partition.

	// At this point we might have an expired cache entry and we know that
	// there was a problem getting the ACL from the ACL datacenter. If a
	// local ACL fault function is registered to query replicated ACL data,
	// and the user's policy allows it, we will try locally before we give
	// up.
	if c.local != nil && c.config.ACLDownPolicy == "extend-cache" {
		parent, rules, err := c.local(id)
		if err != nil {
			// We don't make an exception here for ACLs that aren't
			// found locally. It seems more robust to use an expired
			// cached entry (if we have one) rather than ignore it
			// for the case that replication was a bit behind and
			// didn't have the ACL yet.
			c.logger.Printf("[DEBUG] consul.acl: Failed to get policy from replicated ACLs: %v", err)
			goto ACL_DOWN
		}

		policy, err := acl.Parse(rules)
		if err != nil {
			c.logger.Printf("[DEBUG] consul.acl: Failed to parse policy for replicated ACL: %v", err)
			goto ACL_DOWN
		}
		policy.ID = acl.RuleID(rules)

		// Fake up an ACL datacenter reply and inject it into the cache.
		// Note we use the local TTL here, so this'll be used for that
		// amount of time even once the ACL datacenter becomes available.
		metrics.IncrCounter([]string{"consul", "acl", "replication_hit"}, 1)
		reply.ETag = makeACLETag(parent, policy)
		reply.TTL = c.config.ACLTTL
		reply.Parent = parent
		reply.Policy = policy
		return c.useACLPolicy(id, authDC, cached, &reply)
	}

ACL_DOWN:
	// Unable to refresh, apply the down policy.
	switch c.config.ACLDownPolicy {
	case "allow":
		return acl.AllowAll(), nil
	case "extend-cache":
		if cached != nil {
			return cached.ACL, nil
		}
		fallthrough
	default:
		return acl.DenyAll(), nil
	}
}