df27fa0c84
- Introduce a new telemetry configurable parameter retry_failed_connection. User can set the value to true to let consul agent continue its start process on failed connection to datadog server. When set to false, agent will stop on failed start. The default behavior is true. Co-authored-by: Dan Upton <daniel@floppy.co> Co-authored-by: Evan Culver <eculver@users.noreply.github.com>
523 lines
16 KiB
Go
523 lines
16 KiB
Go
package agent
|
|
|
|
import (
|
|
"fmt"
|
|
"io"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/armon/go-metrics"
|
|
"github.com/hashicorp/go-hclog"
|
|
"github.com/hashicorp/serf/serf"
|
|
|
|
"github.com/hashicorp/consul/acl"
|
|
"github.com/hashicorp/consul/agent/config"
|
|
"github.com/hashicorp/consul/agent/consul"
|
|
"github.com/hashicorp/consul/agent/local"
|
|
"github.com/hashicorp/consul/agent/structs"
|
|
"github.com/hashicorp/consul/api"
|
|
"github.com/hashicorp/consul/lib"
|
|
"github.com/hashicorp/consul/sdk/testutil"
|
|
"github.com/hashicorp/consul/types"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
type authzResolver func(string) (structs.ACLIdentity, acl.Authorizer, error)
|
|
type identResolver func(string) (structs.ACLIdentity, error)
|
|
|
|
type TestACLAgent struct {
|
|
resolveAuthzFn authzResolver
|
|
resolveIdentFn identResolver
|
|
|
|
*Agent
|
|
}
|
|
|
|
// NewTestACLAgent does just enough so that all the code within agent/acl.go can work
|
|
// Basically it needs a local state for some of the vet* functions, a logger and a delegate.
|
|
// The key is that we are the delegate so we can control the ResolveToken responses
|
|
func NewTestACLAgent(t *testing.T, name string, hcl string, resolveAuthz authzResolver, resolveIdent identResolver) *TestACLAgent {
|
|
t.Helper()
|
|
|
|
if resolveIdent == nil {
|
|
resolveIdent = func(s string) (structs.ACLIdentity, error) {
|
|
return nil, nil
|
|
}
|
|
}
|
|
|
|
a := &TestACLAgent{resolveAuthzFn: resolveAuthz, resolveIdentFn: resolveIdent}
|
|
|
|
dataDir := testutil.TempDir(t, "acl-agent")
|
|
|
|
logBuffer := testutil.NewLogBuffer(t)
|
|
loader := func(source config.Source) (config.LoadResult, error) {
|
|
dataDir := fmt.Sprintf(`data_dir = "%s"`, dataDir)
|
|
opts := config.LoadOpts{
|
|
HCL: []string{TestConfigHCL(NodeID()), hcl, dataDir},
|
|
DefaultConfig: source,
|
|
}
|
|
result, err := config.Load(opts)
|
|
if result.RuntimeConfig != nil {
|
|
result.RuntimeConfig.Telemetry.Disable = true
|
|
}
|
|
return result, err
|
|
}
|
|
bd, err := NewBaseDeps(loader, logBuffer)
|
|
require.NoError(t, err)
|
|
|
|
bd.Logger = hclog.NewInterceptLogger(&hclog.LoggerOptions{
|
|
Name: name,
|
|
Level: testutil.TestLogLevel,
|
|
Output: logBuffer,
|
|
TimeFormat: "04:05.000",
|
|
})
|
|
bd.MetricsConfig = &lib.MetricsConfig{
|
|
Handler: metrics.NewInmemSink(1*time.Second, time.Minute),
|
|
}
|
|
|
|
agent, err := New(bd)
|
|
require.NoError(t, err)
|
|
|
|
agent.delegate = a
|
|
agent.State = local.NewState(LocalConfig(bd.RuntimeConfig), bd.Logger, bd.Tokens)
|
|
agent.State.TriggerSyncChanges = func() {}
|
|
a.Agent = agent
|
|
return a
|
|
}
|
|
|
|
func (a *TestACLAgent) ResolveToken(secretID string) (acl.Authorizer, error) {
|
|
if a.resolveAuthzFn == nil {
|
|
return nil, fmt.Errorf("ResolveToken call is unexpected - no authz resolver callback set")
|
|
}
|
|
|
|
_, authz, err := a.resolveAuthzFn(secretID)
|
|
return authz, err
|
|
}
|
|
|
|
func (a *TestACLAgent) ResolveTokenAndDefaultMeta(secretID string, entMeta *acl.EnterpriseMeta, authzContext *acl.AuthorizerContext) (consul.ACLResolveResult, error) {
|
|
authz, err := a.ResolveToken(secretID)
|
|
if err != nil {
|
|
return consul.ACLResolveResult{}, err
|
|
}
|
|
|
|
identity, err := a.resolveIdentFn(secretID)
|
|
if err != nil {
|
|
return consul.ACLResolveResult{}, err
|
|
}
|
|
|
|
// Default the EnterpriseMeta based on the Tokens meta or actual defaults
|
|
// in the case of unknown identity
|
|
if identity != nil {
|
|
entMeta.Merge(identity.EnterpriseMetadata())
|
|
} else {
|
|
entMeta.Merge(structs.DefaultEnterpriseMetaInDefaultPartition())
|
|
}
|
|
|
|
// Use the meta to fill in the ACL authorization context
|
|
entMeta.FillAuthzContext(authzContext)
|
|
|
|
return consul.ACLResolveResult{Authorizer: authz, ACLIdentity: identity}, err
|
|
}
|
|
|
|
// All of these are stubs to satisfy the interface
|
|
func (a *TestACLAgent) GetLANCoordinate() (lib.CoordinateSet, error) {
|
|
return nil, fmt.Errorf("Unimplemented")
|
|
}
|
|
func (a *TestACLAgent) Leave() error {
|
|
return fmt.Errorf("Unimplemented")
|
|
}
|
|
func (a *TestACLAgent) LANMembersInAgentPartition() []serf.Member {
|
|
return nil
|
|
}
|
|
func (a *TestACLAgent) LANMembers(f consul.LANMemberFilter) ([]serf.Member, error) {
|
|
return nil, fmt.Errorf("Unimplemented")
|
|
}
|
|
func (a *TestACLAgent) AgentLocalMember() serf.Member {
|
|
return serf.Member{}
|
|
}
|
|
func (a *TestACLAgent) JoinLAN(addrs []string, entMeta *acl.EnterpriseMeta) (n int, err error) {
|
|
return 0, fmt.Errorf("Unimplemented")
|
|
}
|
|
func (a *TestACLAgent) RemoveFailedNode(node string, prune bool, entMeta *acl.EnterpriseMeta) error {
|
|
return fmt.Errorf("Unimplemented")
|
|
}
|
|
func (a *TestACLAgent) RPC(method string, args interface{}, reply interface{}) error {
|
|
return fmt.Errorf("Unimplemented")
|
|
}
|
|
func (a *TestACLAgent) SnapshotRPC(args *structs.SnapshotRequest, in io.Reader, out io.Writer, replyFn structs.SnapshotReplyFn) error {
|
|
return fmt.Errorf("Unimplemented")
|
|
}
|
|
func (a *TestACLAgent) Shutdown() error {
|
|
return fmt.Errorf("Unimplemented")
|
|
}
|
|
func (a *TestACLAgent) Stats() map[string]map[string]string {
|
|
return nil
|
|
}
|
|
func (a *TestACLAgent) ReloadConfig(_ consul.ReloadableConfig) error {
|
|
return fmt.Errorf("Unimplemented")
|
|
}
|
|
|
|
func TestACL_Version8EnabledByDefault(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
called := false
|
|
resolveFn := func(string) (structs.ACLIdentity, acl.Authorizer, error) {
|
|
called = true
|
|
return nil, nil, acl.ErrNotFound
|
|
}
|
|
a := NewTestACLAgent(t, t.Name(), TestACLConfig(), resolveFn, nil)
|
|
|
|
_, err := a.delegate.ResolveTokenAndDefaultMeta("nope", nil, nil)
|
|
require.Error(t, err)
|
|
require.True(t, called)
|
|
}
|
|
|
|
func authzFromPolicy(policy *acl.Policy, cfg *acl.Config) (acl.Authorizer, error) {
|
|
return acl.NewPolicyAuthorizerWithDefaults(acl.DenyAll(), []*acl.Policy{policy}, cfg)
|
|
}
|
|
|
|
type testToken struct {
|
|
token structs.ACLToken
|
|
// yes the rules can exist on the token itself but that is legacy behavior
|
|
// that I would prefer these tests not rely on
|
|
rules string
|
|
}
|
|
|
|
var (
|
|
nodeROSecret = "7e80d017-bccc-492f-8dec-65f03aeaebf3"
|
|
nodeRWSecret = "e3586ee5-02a2-4bf4-9ec3-9c4be7606e8c"
|
|
serviceROSecret = "3d2c8552-df3b-4da7-9890-36885cbf56ac"
|
|
serviceRWSecret = "4a1017a2-f788-4be3-93f2-90566f1340bb"
|
|
otherRWSecret = "a38e8016-91b6-4876-b3e7-a307abbb2002"
|
|
|
|
testTokens = map[string]testToken{
|
|
nodeROSecret: {
|
|
token: structs.ACLToken{
|
|
AccessorID: "9df2d1a4-2d07-414e-8ead-6053f56ed2eb",
|
|
SecretID: nodeROSecret,
|
|
},
|
|
rules: `node_prefix "Node" { policy = "read" }`,
|
|
},
|
|
nodeRWSecret: {
|
|
token: structs.ACLToken{
|
|
AccessorID: "efb6b7d5-d343-47c1-b4cb-aa6b94d2f490",
|
|
SecretID: nodeRWSecret,
|
|
},
|
|
rules: `node_prefix "Node" { policy = "write" }`,
|
|
},
|
|
serviceROSecret: {
|
|
token: structs.ACLToken{
|
|
AccessorID: "0da53edb-36e5-4603-9c31-79965bad45f5",
|
|
SecretID: serviceROSecret,
|
|
},
|
|
rules: `service_prefix "service" { policy = "read" }`,
|
|
},
|
|
serviceRWSecret: {
|
|
token: structs.ACLToken{
|
|
AccessorID: "52504258-137a-41e6-9326-01f40e80872e",
|
|
SecretID: serviceRWSecret,
|
|
},
|
|
rules: `service_prefix "service" { policy = "write" }`,
|
|
},
|
|
otherRWSecret: {
|
|
token: structs.ACLToken{
|
|
AccessorID: "5e032c5b-c39e-4552-b5ad-8a9365b099c4",
|
|
SecretID: otherRWSecret,
|
|
},
|
|
rules: `service_prefix "other" { policy = "write" }`,
|
|
},
|
|
}
|
|
)
|
|
|
|
func catalogPolicy(token string) (structs.ACLIdentity, acl.Authorizer, error) {
|
|
tok, ok := testTokens[token]
|
|
if !ok {
|
|
return nil, nil, acl.ErrNotFound
|
|
}
|
|
|
|
policy, err := acl.NewPolicyFromSource(tok.rules, acl.SyntaxCurrent, nil, nil)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
authz, err := authzFromPolicy(policy, nil)
|
|
return &tok.token, authz, err
|
|
}
|
|
|
|
func catalogIdent(token string) (structs.ACLIdentity, error) {
|
|
tok, ok := testTokens[token]
|
|
if !ok {
|
|
return nil, acl.ErrNotFound
|
|
}
|
|
|
|
return &tok.token, nil
|
|
}
|
|
|
|
func TestACL_vetServiceRegister(t *testing.T) {
|
|
t.Parallel()
|
|
a := NewTestACLAgent(t, t.Name(), TestACLConfig(), catalogPolicy, catalogIdent)
|
|
|
|
// Register a new service, with permission.
|
|
err := a.vetServiceRegister(serviceRWSecret, &structs.NodeService{
|
|
ID: "my-service",
|
|
Service: "service",
|
|
})
|
|
require.NoError(t, err)
|
|
|
|
// Register a new service without write privs.
|
|
err = a.vetServiceRegister(serviceROSecret, &structs.NodeService{
|
|
ID: "my-service",
|
|
Service: "service",
|
|
})
|
|
require.True(t, acl.IsErrPermissionDenied(err))
|
|
|
|
// Try to register over a service without write privs to the existing
|
|
// service.
|
|
a.State.AddService(&structs.NodeService{
|
|
ID: "my-service",
|
|
Service: "other",
|
|
}, "")
|
|
err = a.vetServiceRegister(serviceRWSecret, &structs.NodeService{
|
|
ID: "my-service",
|
|
Service: "service",
|
|
})
|
|
require.True(t, acl.IsErrPermissionDenied(err))
|
|
}
|
|
|
|
func TestACL_vetServiceUpdateWithAuthorizer(t *testing.T) {
|
|
t.Parallel()
|
|
a := NewTestACLAgent(t, t.Name(), TestACLConfig(), catalogPolicy, catalogIdent)
|
|
|
|
vetServiceUpdate := func(token string, serviceID structs.ServiceID) error {
|
|
authz, err := a.delegate.ResolveTokenAndDefaultMeta(token, nil, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return a.vetServiceUpdateWithAuthorizer(authz, serviceID)
|
|
}
|
|
|
|
// Update a service that doesn't exist.
|
|
err := vetServiceUpdate(serviceRWSecret, structs.NewServiceID("my-service", nil))
|
|
require.Error(t, err)
|
|
require.Contains(t, err.Error(), "Unknown service")
|
|
|
|
// Update with write privs.
|
|
a.State.AddService(&structs.NodeService{
|
|
ID: "my-service",
|
|
Service: "service",
|
|
}, "")
|
|
err = vetServiceUpdate(serviceRWSecret, structs.NewServiceID("my-service", nil))
|
|
require.NoError(t, err)
|
|
|
|
// Update without write privs.
|
|
err = vetServiceUpdate(serviceROSecret, structs.NewServiceID("my-service", nil))
|
|
require.Error(t, err)
|
|
require.True(t, acl.IsErrPermissionDenied(err))
|
|
}
|
|
|
|
func TestACL_vetCheckRegisterWithAuthorizer(t *testing.T) {
|
|
t.Parallel()
|
|
a := NewTestACLAgent(t, t.Name(), TestACLConfig(), catalogPolicy, catalogIdent)
|
|
|
|
vetCheckRegister := func(token string, check *structs.HealthCheck) error {
|
|
authz, err := a.delegate.ResolveTokenAndDefaultMeta(token, nil, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return a.vetCheckRegisterWithAuthorizer(authz, check)
|
|
}
|
|
|
|
// Register a new service check with write privs.
|
|
err := vetCheckRegister(serviceRWSecret, &structs.HealthCheck{
|
|
CheckID: types.CheckID("my-check"),
|
|
ServiceID: "my-service",
|
|
ServiceName: "service",
|
|
})
|
|
require.NoError(t, err)
|
|
|
|
// Register a new service check without write privs.
|
|
err = vetCheckRegister(serviceROSecret, &structs.HealthCheck{
|
|
CheckID: types.CheckID("my-check"),
|
|
ServiceID: "my-service",
|
|
ServiceName: "service",
|
|
})
|
|
require.Error(t, err)
|
|
require.True(t, acl.IsErrPermissionDenied(err))
|
|
|
|
// Register a new node check with write privs.
|
|
err = vetCheckRegister(nodeRWSecret, &structs.HealthCheck{
|
|
CheckID: types.CheckID("my-check"),
|
|
})
|
|
require.NoError(t, err)
|
|
|
|
// Register a new node check without write privs.
|
|
err = vetCheckRegister(nodeROSecret, &structs.HealthCheck{
|
|
CheckID: types.CheckID("my-check"),
|
|
})
|
|
require.Error(t, err)
|
|
require.True(t, acl.IsErrPermissionDenied(err))
|
|
|
|
// Try to register over a service check without write privs to the
|
|
// existing service.
|
|
a.State.AddService(&structs.NodeService{
|
|
ID: "my-service",
|
|
Service: "service",
|
|
}, "")
|
|
a.State.AddCheck(&structs.HealthCheck{
|
|
CheckID: types.CheckID("my-check"),
|
|
ServiceID: "my-service",
|
|
ServiceName: "other",
|
|
}, "")
|
|
err = vetCheckRegister(serviceRWSecret, &structs.HealthCheck{
|
|
CheckID: types.CheckID("my-check"),
|
|
ServiceID: "my-service",
|
|
ServiceName: "service",
|
|
})
|
|
require.Error(t, err)
|
|
require.True(t, acl.IsErrPermissionDenied(err))
|
|
|
|
// Try to register over a node check without write privs to the node.
|
|
a.State.AddCheck(&structs.HealthCheck{
|
|
CheckID: types.CheckID("my-node-check"),
|
|
}, "")
|
|
err = vetCheckRegister(serviceRWSecret, &structs.HealthCheck{
|
|
CheckID: types.CheckID("my-node-check"),
|
|
ServiceID: "my-service",
|
|
ServiceName: "service",
|
|
})
|
|
require.Error(t, err)
|
|
require.True(t, acl.IsErrPermissionDenied(err))
|
|
}
|
|
|
|
func TestACL_vetCheckUpdateWithAuthorizer(t *testing.T) {
|
|
t.Parallel()
|
|
a := NewTestACLAgent(t, t.Name(), TestACLConfig(), catalogPolicy, catalogIdent)
|
|
|
|
vetCheckUpdate := func(token string, checkID structs.CheckID) error {
|
|
authz, err := a.delegate.ResolveTokenAndDefaultMeta(token, nil, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return a.vetCheckUpdateWithAuthorizer(authz, checkID)
|
|
}
|
|
|
|
// Update a check that doesn't exist.
|
|
err := vetCheckUpdate(nodeRWSecret, structs.NewCheckID("my-check", nil))
|
|
require.Error(t, err)
|
|
require.Contains(t, err.Error(), "Unknown check")
|
|
|
|
// Update service check with write privs.
|
|
a.State.AddService(&structs.NodeService{
|
|
ID: "my-service",
|
|
Service: "service",
|
|
}, "")
|
|
a.State.AddCheck(&structs.HealthCheck{
|
|
CheckID: types.CheckID("my-service-check"),
|
|
ServiceID: "my-service",
|
|
ServiceName: "service",
|
|
}, "")
|
|
err = vetCheckUpdate(serviceRWSecret, structs.NewCheckID("my-service-check", nil))
|
|
require.NoError(t, err)
|
|
|
|
// Update service check without write privs.
|
|
err = vetCheckUpdate(serviceROSecret, structs.NewCheckID("my-service-check", nil))
|
|
require.Error(t, err)
|
|
require.True(t, acl.IsErrPermissionDenied(err), "not permission denied: %s", err.Error())
|
|
|
|
// Update node check with write privs.
|
|
a.State.AddCheck(&structs.HealthCheck{
|
|
CheckID: types.CheckID("my-node-check"),
|
|
}, "")
|
|
err = vetCheckUpdate(nodeRWSecret, structs.NewCheckID("my-node-check", nil))
|
|
require.NoError(t, err)
|
|
|
|
// Update without write privs.
|
|
err = vetCheckUpdate(nodeROSecret, structs.NewCheckID("my-node-check", nil))
|
|
require.Error(t, err)
|
|
require.True(t, acl.IsErrPermissionDenied(err))
|
|
}
|
|
|
|
func TestACL_filterMembers(t *testing.T) {
|
|
t.Parallel()
|
|
a := NewTestACLAgent(t, t.Name(), TestACLConfig(), catalogPolicy, catalogIdent)
|
|
|
|
var members []serf.Member
|
|
require.NoError(t, a.filterMembers(nodeROSecret, &members))
|
|
require.Len(t, members, 0)
|
|
|
|
members = []serf.Member{
|
|
{Name: "Node 1"},
|
|
{Name: "Nope"},
|
|
{Name: "Node 2"},
|
|
}
|
|
require.NoError(t, a.filterMembers(nodeROSecret, &members))
|
|
require.Len(t, members, 2)
|
|
require.Equal(t, members[0].Name, "Node 1")
|
|
require.Equal(t, members[1].Name, "Node 2")
|
|
}
|
|
|
|
func TestACL_filterServicesWithAuthorizer(t *testing.T) {
|
|
t.Parallel()
|
|
a := NewTestACLAgent(t, t.Name(), TestACLConfig(), catalogPolicy, catalogIdent)
|
|
|
|
filterServices := func(token string, services map[string]*api.AgentService) error {
|
|
authz, err := a.delegate.ResolveTokenAndDefaultMeta(token, nil, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return a.filterServicesWithAuthorizer(authz, services)
|
|
}
|
|
|
|
services := make(map[string]*api.AgentService)
|
|
require.NoError(t, filterServices(nodeROSecret, services))
|
|
|
|
services[structs.NewServiceID("my-service", nil).String()] = &api.AgentService{ID: "my-service", Service: "service"}
|
|
services[structs.NewServiceID("my-other", nil).String()] = &api.AgentService{ID: "my-other", Service: "other"}
|
|
require.NoError(t, filterServices(serviceROSecret, services))
|
|
|
|
require.Contains(t, services, structs.NewServiceID("my-service", nil).String())
|
|
require.NotContains(t, services, structs.NewServiceID("my-other", nil).String())
|
|
}
|
|
|
|
func TestACL_filterChecksWithAuthorizer(t *testing.T) {
|
|
t.Parallel()
|
|
a := NewTestACLAgent(t, t.Name(), TestACLConfig(), catalogPolicy, catalogIdent)
|
|
|
|
filterChecks := func(token string, checks map[types.CheckID]*structs.HealthCheck) error {
|
|
authz, err := a.delegate.ResolveTokenAndDefaultMeta(token, nil, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return a.filterChecksWithAuthorizer(authz, checks)
|
|
}
|
|
|
|
checks := make(map[types.CheckID]*structs.HealthCheck)
|
|
require.NoError(t, filterChecks(nodeROSecret, checks))
|
|
|
|
checks["my-node"] = &structs.HealthCheck{}
|
|
checks["my-service"] = &structs.HealthCheck{ServiceName: "service"}
|
|
checks["my-other"] = &structs.HealthCheck{ServiceName: "other"}
|
|
require.NoError(t, filterChecks(serviceROSecret, checks))
|
|
_, ok := checks["my-node"]
|
|
require.False(t, ok)
|
|
_, ok = checks["my-service"]
|
|
require.True(t, ok)
|
|
_, ok = checks["my-other"]
|
|
require.False(t, ok)
|
|
|
|
checks["my-node"] = &structs.HealthCheck{}
|
|
checks["my-service"] = &structs.HealthCheck{ServiceName: "service"}
|
|
checks["my-other"] = &structs.HealthCheck{ServiceName: "other"}
|
|
require.NoError(t, filterChecks(nodeROSecret, checks))
|
|
_, ok = checks["my-node"]
|
|
require.True(t, ok)
|
|
_, ok = checks["my-service"]
|
|
require.False(t, ok)
|
|
_, ok = checks["my-other"]
|
|
require.False(t, ok)
|
|
}
|