status-im
/
consul
mirror of https://github.com/status-im/consul.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity
consul/agent/xds/server_test.go

1185 lines
35 KiB
Go
Raw Blame History

package xds
import (
"bytes"
"context"
"errors"
"fmt"
"log"
"os"
"strings"
"sync"
"sync/atomic"
"testing"
"text/template"
"time"
envoy "github.com/envoyproxy/go-control-plane/envoy/api/v2"
"github.com/gogo/protobuf/jsonpb"
"github.com/stretchr/testify/require"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
"github.com/hashicorp/consul/acl"
"github.com/hashicorp/consul/agent/cache"
"github.com/hashicorp/consul/agent/proxycfg"
"github.com/hashicorp/consul/agent/structs"
)
// testManager is a mock of proxycfg.Manager that's simpler to control for
// testing. It also implements ConnectAuthz to allow control over authorization.
type testManager struct {
sync.Mutex
chans map[string]chan *proxycfg.ConfigSnapshot
cancels chan string
authz map[string]connectAuthzResult
}
type connectAuthzResult struct {
authz bool
reason string
m *cache.ResultMeta
err error
}
func newTestManager(t *testing.T) *testManager {
return &testManager{
chans: map[string]chan *proxycfg.ConfigSnapshot{},
cancels: make(chan string, 10),
authz: make(map[string]connectAuthzResult),
}
}
// RegisterProxy simulates a proxy registration
func (m *testManager) RegisterProxy(t *testing.T, proxyID string) {
m.Lock()
defer m.Unlock()
m.chans[proxyID] = make(chan *proxycfg.ConfigSnapshot, 1)
}
// Deliver simulates a proxy registration
func (m *testManager) DeliverConfig(t *testing.T, proxyID string, cfg *proxycfg.ConfigSnapshot) {
t.Helper()
m.Lock()
defer m.Unlock()
select {
case m.chans[proxyID] <- cfg:
case <-time.After(10 * time.Millisecond):
t.Fatalf("took too long to deliver config")
}
}
// Watch implements ConfigManager
func (m *testManager) Watch(proxyID string) (<-chan *proxycfg.ConfigSnapshot, proxycfg.CancelFunc) {
m.Lock()
defer m.Unlock()
// ch might be nil but then it will just block forever
return m.chans[proxyID], func() {
m.cancels <- proxyID
}
}
// AssertWatchCancelled checks that the most recent call to a Watch cancel func
// was from the specified proxyID and that one is made in a short time. This
// probably won't work if you are running multiple Watches in parallel on
// multiple proxyIDS due to timing/ordering issues but I don't think we need to
// do that.
func (m *testManager) AssertWatchCancelled(t *testing.T, proxyID string) {
t.Helper()
select {
case got := <-m.cancels:
require.Equal(t, proxyID, got)
case <-time.After(50 * time.Millisecond):
t.Fatalf("timed out waiting for Watch cancel for %s", proxyID)
}
}
// ConnectAuthorize implements ConnectAuthz
func (m *testManager) ConnectAuthorize(token string, req *structs.ConnectAuthorizeRequest) (authz bool, reason string, meta *cache.ResultMeta, err error) {
m.Lock()
defer m.Unlock()
if res, ok := m.authz[token]; ok {
return res.authz, res.reason, res.m, res.err
}
// Default allow but with reason that won't match by accident in a test case
return true, "OK: allowed by default test implementation", nil, nil
}
func TestServer_StreamAggregatedResources_BasicProtocol(t *testing.T) {
logger := log.New(os.Stderr, "", log.LstdFlags)
mgr := newTestManager(t)
aclResolve := func(id string) (acl.Authorizer, error) {
// Allow all
return acl.RootAuthorizer("manage"), nil
}
envoy := NewTestEnvoy(t, "web-sidecar-proxy", "")
defer envoy.Close()
s := Server{
Logger: logger,
CfgMgr: mgr,
Authz: mgr,
ResolveToken: aclResolve,
}
s.Initialize()
go func() {
err := s.StreamAggregatedResources(envoy.stream)
require.NoError(t, err)
}()
// Register the proxy to create state needed to Watch() on
mgr.RegisterProxy(t, "web-sidecar-proxy")
// Send initial cluster discover
envoy.SendReq(t, ClusterType, 0, 0)
// Check no response sent yet
assertChanBlocked(t, envoy.stream.sendCh)
// Deliver a new snapshot
snap := proxycfg.TestConfigSnapshot(t)
mgr.DeliverConfig(t, "web-sidecar-proxy", snap)
assertResponseSent(t, envoy.stream.sendCh, expectClustersJSON(t, snap, "", 1, 1))
// Envoy then tries to discover endpoints for those clusters. Technically it
// includes the cluster names in the ResourceNames field but we ignore that
// completely for now so not bothering to simulate that.
envoy.SendReq(t, EndpointType, 0, 0)
// It also (in parallel) issues the next cluster request (which acts as an ACK
// of the version we sent)
envoy.SendReq(t, ClusterType, 1, 1)
// We should get a response immediately since the config is already present in
// the server for endpoints. Note that this should not be racy if the server
// is behaving well since the Cluster send above should be blocked until we
// deliver a new config version.
assertResponseSent(t, envoy.stream.sendCh, expectEndpointsJSON(t, snap, "", 1, 2))
// And no other response yet
assertChanBlocked(t, envoy.stream.sendCh)
// Envoy now sends listener request along with next endpoint one
envoy.SendReq(t, ListenerType, 0, 0)
envoy.SendReq(t, EndpointType, 1, 2)
// And should get a response immediately.
assertResponseSent(t, envoy.stream.sendCh, expectListenerJSON(t, snap, "", 1, 3))
// Now send Route request along with next listener one
envoy.SendReq(t, RouteType, 0, 0)
envoy.SendReq(t, ListenerType, 1, 3)
// We don't serve routes yet so this should block with no response
assertChanBlocked(t, envoy.stream.sendCh)
// WOOP! Envoy now has full connect config. Lets verify that if we update it,
// all the responses get resent with the new version. We don't actually want
// to change everything because that's tedious - our implementation will
// actually resend all blocked types on the new "version" anyway since it
// doesn't know _what_ changed. We could do something trivial but let's
// simulate a leaf cert expiring and being rotated.
snap.Leaf = proxycfg.TestLeafForCA(t, snap.Roots.Roots[0])
mgr.DeliverConfig(t, "web-sidecar-proxy", snap)
// All 3 response that have something to return should return with new version
// note that the ordering is not deterministic in general. Trying to make this
// test order-agnostic though is a massive pain since we are comparing
// non-identical JSON strings (so can simply sort by anything) and because we
// don't know the order the nonces will be assigned. For now we rely and
// require our implementation to always deliver updates in a specific order
// which is reasonable anyway to ensure consistency of the config Envoy sees.
assertResponseSent(t, envoy.stream.sendCh, expectClustersJSON(t, snap, "", 2, 4))
assertResponseSent(t, envoy.stream.sendCh, expectEndpointsJSON(t, snap, "", 2, 5))
assertResponseSent(t, envoy.stream.sendCh, expectListenerJSON(t, snap, "", 2, 6))
// Let's pretend that Envoy doesn't like that new listener config. It will ACK
// all the others (same version) but NACK the listener. This is the most
// subtle part of xDS and the server implementation so I'll elaborate. A full
// description of the protocol can be found at
// https://github.com/envoyproxy/data-plane-api/blob/master/XDS_PROTOCOL.md.
// Envoy delays making a followup request for a type until after it has
// processed and applied the last response. The next request then will include
// the nonce in the last response which acknowledges _receiving_ and handling
// that response. It also includes the currently applied version. If all is
// good and it successfully applies the config, then the version in the next
// response will be the same version just sent. This is considered to be an
// ACK of that version for that type. If envoy fails to apply the config for
// some reason, it will still acknowledge that it received it (still return
// the responses nonce), but will show the previous version it's still using.
// This is considered a NACK. It's important that the server pay attention to
// the _nonce_ and not the version when deciding what to send otherwise a bad
// version that can't be applied in Envoy will cause a busy loop.
//
// In this case we are simulating that Envoy failed to apply the Listener
// response but did apply the other types so all get the new nonces, but
// listener stays on v1.
envoy.SendReq(t, ClusterType, 2, 4)
envoy.SendReq(t, EndpointType, 2, 5)
envoy.SendReq(t, ListenerType, 1, 6) // v1 is a NACK
// Even though we nacked, we should still NOT get then v2 listeners
// redelivered since nothing has changed.
assertChanBlocked(t, envoy.stream.sendCh)
// Change config again and make sure it's delivered to everyone!
snap.Leaf = proxycfg.TestLeafForCA(t, snap.Roots.Roots[0])
mgr.DeliverConfig(t, "web-sidecar-proxy", snap)
assertResponseSent(t, envoy.stream.sendCh, expectClustersJSON(t, snap, "", 3, 7))
assertResponseSent(t, envoy.stream.sendCh, expectEndpointsJSON(t, snap, "", 3, 8))
assertResponseSent(t, envoy.stream.sendCh, expectListenerJSON(t, snap, "", 3, 9))
}
func expectListenerJSONResources(t *testing.T, snap *proxycfg.ConfigSnapshot, token string, v, n uint64) map[string]string {
tokenVal := ""
if token != "" {
tokenVal = fmt.Sprintf(",\n"+`"value": "%s"`, token)
}
return map[string]string{
"public_listener": `{
"@type": "type.googleapis.com/envoy.api.v2.Listener",
"name": "public_listener:0.0.0.0:9999",
"address": {
"socketAddress": {
"address": "0.0.0.0",
"portValue": 9999
}
},
"filterChains": [
{
"tlsContext": ` + expectedPublicTLSContextJSON(t, snap) + `,
"filters": [
{
"name": "envoy.ext_authz",
"config": {
"grpc_service": {
"envoy_grpc": {
"cluster_name": "local_agent"
},
"initial_metadata": [
{
"key": "x-consul-token"
` + tokenVal + `
}
]
},
"stat_prefix": "connect_authz"
}
},
{
"name": "envoy.tcp_proxy",
"config": {
"cluster": "local_app",
"stat_prefix": "public_listener"
}
}
]
}
]
}`,
"service:db": `{
"@type": "type.googleapis.com/envoy.api.v2.Listener",
"name": "service:db:127.0.0.1:9191",
"address": {
"socketAddress": {
"address": "127.0.0.1",
"portValue": 9191
}
},
"filterChains": [
{
"filters": [
{
"name": "envoy.tcp_proxy",
"config": {
"cluster": "service:db",
"stat_prefix": "service:db"
}
}
]
}
]
}`,
"prepared_query:geo-cache": `{
"@type": "type.googleapis.com/envoy.api.v2.Listener",
"name": "prepared_query:geo-cache:127.10.10.10:8181",
"address": {
"socketAddress": {
"address": "127.10.10.10",
"portValue": 8181
}
},
"filterChains": [
{
"filters": [
{
"name": "envoy.tcp_proxy",
"config": {
"cluster": "prepared_query:geo-cache",
"stat_prefix": "prepared_query:geo-cache"
}
}
]
}
]
}`,
}
}
func expectListenerJSONFromResources(t *testing.T, snap *proxycfg.ConfigSnapshot, token string, v, n uint64, resourcesJSON map[string]string) string {
resJSON := ""
// Sort resources into specific order because that matters in JSONEq
// comparison later.
keyOrder := []string{"public_listener"}
for _, u := range snap.Proxy.Upstreams {
keyOrder = append(keyOrder, u.Identifier())
}
for _, k := range keyOrder {
j, ok := resourcesJSON[k]
if !ok {
continue
}
if resJSON != "" {
resJSON += ",\n"
}
resJSON += j
}
return `{
"versionInfo": "` + hexString(v) + `",
"resources": [` + resJSON + `],
"typeUrl": "type.googleapis.com/envoy.api.v2.Listener",
"nonce": "` + hexString(n) + `"
}`
}
func expectListenerJSON(t *testing.T, snap *proxycfg.ConfigSnapshot, token string, v, n uint64) string {
return expectListenerJSONFromResources(t, snap, token, v, n,
expectListenerJSONResources(t, snap, token, v, n))
}
func expectEndpointsJSON(t *testing.T, snap *proxycfg.ConfigSnapshot, token string, v, n uint64) string {
return `{
"versionInfo": "` + hexString(v) + `",
"resources": [
{
"@type": "type.googleapis.com/envoy.api.v2.ClusterLoadAssignment",
"clusterName": "service:db",
"endpoints": [
{
"lbEndpoints": [
{
"endpoint": {
"address": {
"socketAddress": {
"address": "10.10.1.1",
"portValue": 0
}
}
}
},
{
"endpoint": {
"address": {
"socketAddress": {
"address": "10.10.1.2",
"portValue": 0
}
}
}
}
]
}
]
}
],
"typeUrl": "type.googleapis.com/envoy.api.v2.ClusterLoadAssignment",
"nonce": "` + hexString(n) + `"
}`
}
func expectClustersJSON(t *testing.T, snap *proxycfg.ConfigSnapshot, token string, v, n uint64) string {
return `{
"versionInfo": "` + hexString(v) + `",
"resources": [
{
"@type": "type.googleapis.com/envoy.api.v2.Cluster",
"name": "local_app",
"connectTimeout": "5s",
"hosts": [
{
"socketAddress": {
"address": "127.0.0.1",
"portValue": 8080
}
}
]
},
{
"@type": "type.googleapis.com/envoy.api.v2.Cluster",
"name": "service:db",
"type": "EDS",
"edsClusterConfig": {
"edsConfig": {
"ads": {
}
}
},
"connectTimeout": "5s",
"tlsContext": ` + expectedUpstreamTLSContextJSON(t, snap) + `
},
{
"@type": "type.googleapis.com/envoy.api.v2.Cluster",
"name": "prepared_query:geo-cache",
"type": "EDS",
"edsClusterConfig": {
"edsConfig": {
"ads": {
}
}
},
"connectTimeout": "5s",
"tlsContext": ` + expectedUpstreamTLSContextJSON(t, snap) + `
}
],
"typeUrl": "type.googleapis.com/envoy.api.v2.Cluster",
"nonce": "` + hexString(n) + `"
}
`
}
func expectedUpstreamTLSContextJSON(t *testing.T, snap *proxycfg.ConfigSnapshot) string {
return expectedTLSContextJSON(t, snap, false)
}
func expectedPublicTLSContextJSON(t *testing.T, snap *proxycfg.ConfigSnapshot) string {
return expectedTLSContextJSON(t, snap, true)
}
func expectedTLSContextJSON(t *testing.T, snap *proxycfg.ConfigSnapshot, requireClientCert bool) string {
// Assume just one root for now, can get fancier later if needed.
caPEM := snap.Roots.Roots[0].RootCert
reqClient := ""
if requireClientCert {
reqClient = `,
"requireClientCertificate": true`
}
return `{
"commonTlsContext": {
"tlsParams": {},
"tlsCertificates": [
{
"certificateChain": {
"inlineString": "` + strings.Replace(snap.Leaf.CertPEM, "\n", "\\n", -1) + `"
},
"privateKey": {
"inlineString": "` + strings.Replace(snap.Leaf.PrivateKeyPEM, "\n", "\\n", -1) + `"
}
}
],
"validationContext": {
"trustedCa": {
"inlineString": "` + strings.Replace(caPEM, "\n", "\\n", -1) + `"
}
}
}
` + reqClient + `
}`
}
func assertChanBlocked(t *testing.T, ch chan *envoy.DiscoveryResponse) {
t.Helper()
select {
case r := <-ch:
t.Fatalf("chan should block but received: %v", r)
case <-time.After(10 * time.Millisecond):
return
}
}
func assertResponseSent(t *testing.T, ch chan *envoy.DiscoveryResponse, wantJSON string) {
t.Helper()
select {
case r := <-ch:
assertResponse(t, r, wantJSON)
case <-time.After(50 * time.Millisecond):
t.Fatalf("no response received after 50ms")
}
}
// assertResponse is a helper to test a envoy.DiscoveryResponse matches the
// JSON representation we expect. We use JSON because the responses use protobuf
// Any type which includes binary protobuf encoding and would make creating
// expected structs require the same code that is under test!
func assertResponse(t *testing.T, r *envoy.DiscoveryResponse, wantJSON string) {
t.Helper()
m := jsonpb.Marshaler{
Indent: " ",
}
gotJSON, err := m.MarshalToString(r)
require.NoError(t, err)
require.JSONEqf(t, wantJSON, gotJSON, "got:\n%s", gotJSON)
}
func TestServer_StreamAggregatedResources_ACLEnforcement(t *testing.T) {
tests := []struct {
name string
defaultDeny bool
acl string
token string
wantDenied bool
}{
// Note that although we've stubbed actual ACL checks in the testManager
// ConnectAuthorize mock, by asserting against specific reason strings here
// even in the happy case which can't match the default one returned by the
// mock we are implicitly validating that the implementation used the
// correct token from the context.
{
name: "no ACLs configured",
defaultDeny: false,
wantDenied: false,
},
{
name: "default deny, no token",
defaultDeny: true,
wantDenied: true,
},
{
name: "default deny, service:write token",
defaultDeny: true,
acl: `service "web" { policy = "write" }`,
token: "service-write-on-web",
wantDenied: false,
},
{
name: "default deny, service:read token",
defaultDeny: true,
acl: `service "web" { policy = "read" }`,
token: "service-write-on-web",
wantDenied: true,
},
{
name: "default deny, service:write token on different service",
defaultDeny: true,
acl: `service "not-web" { policy = "write" }`,
token: "service-write-on-not-web",
wantDenied: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
logger := log.New(os.Stderr, "", log.LstdFlags)
mgr := newTestManager(t)
aclResolve := func(id string) (acl.Authorizer, error) {
if !tt.defaultDeny {
// Allow all
return acl.RootAuthorizer("allow"), nil
}
if tt.acl == "" {
// No token and defaultDeny is denied
return acl.RootAuthorizer("deny"), nil
}
// Ensure the correct token was passed
require.Equal(t, tt.token, id)
// Parse the ACL and enforce it
policy, err := acl.NewPolicyFromSource("", 0, tt.acl, acl.SyntaxLegacy, nil)
require.NoError(t, err)
return acl.NewPolicyAuthorizer(acl.RootAuthorizer("deny"), []*acl.Policy{policy}, nil)
}
envoy := NewTestEnvoy(t, "web-sidecar-proxy", tt.token)
defer envoy.Close()
s := Server{
Logger: logger,
CfgMgr: mgr,
Authz: mgr,
ResolveToken: aclResolve,
}
s.Initialize()
errCh := make(chan error, 1)
go func() {
errCh <- s.StreamAggregatedResources(envoy.stream)
}()
// Register the proxy to create state needed to Watch() on
mgr.RegisterProxy(t, "web-sidecar-proxy")
// Deliver a new snapshot
snap := proxycfg.TestConfigSnapshot(t)
mgr.DeliverConfig(t, "web-sidecar-proxy", snap)
// Send initial listener discover, in real life Envoy always sends cluster
// first but it doesn't really matter and listener has a response that
// includes the token in the ext authz filter so lets us test more stuff.
envoy.SendReq(t, ListenerType, 0, 0)
if !tt.wantDenied {
assertResponseSent(t, envoy.stream.sendCh, expectListenerJSON(t, snap, tt.token, 1, 1))
// Close the client stream since all is well. We _don't_ do this in the
// expected error case because we want to verify the error closes the
// stream from server side.
envoy.Close()
}
select {
case err := <-errCh:
if tt.wantDenied {
require.Error(t, err)
require.Contains(t, err.Error(), "permission denied")
mgr.AssertWatchCancelled(t, "web-sidecar-proxy")
} else {
require.NoError(t, err)
}
case <-time.After(50 * time.Millisecond):
t.Fatalf("timed out waiting for handler to finish")
}
})
}
}
func TestServer_StreamAggregatedResources_ACLTokenDeleted_StreamTerminatedDuringDiscoveryRequest(t *testing.T) {
aclRules := `service "web" { policy = "write" }`
token := "service-write-on-web"
policy, err := acl.NewPolicyFromSource("", 0, aclRules, acl.SyntaxLegacy, nil)
require.NoError(t, err)
var validToken atomic.Value
validToken.Store(token)
logger := log.New(os.Stderr, "", log.LstdFlags)
mgr := newTestManager(t)
aclResolve := func(id string) (acl.Authorizer, error) {
if token := validToken.Load(); token == nil || id != token.(string) {
return nil, acl.ErrNotFound
}
return acl.NewPolicyAuthorizer(acl.RootAuthorizer("deny"), []*acl.Policy{policy}, nil)
}
envoy := NewTestEnvoy(t, "web-sidecar-proxy", token)
defer envoy.Close()
s := Server{
Logger: logger,
CfgMgr: mgr,
Authz: mgr,
ResolveToken: aclResolve,
AuthCheckFrequency: 1 * time.Hour, // make sure this doesn't kick in
}
s.Initialize()
errCh := make(chan error, 1)
go func() {
errCh <- s.StreamAggregatedResources(envoy.stream)
}()
getError := func() (gotErr error, ok bool) {
select {
case err := <-errCh:
return err, true
default:
return nil, false
}
}
// Register the proxy to create state needed to Watch() on
mgr.RegisterProxy(t, "web-sidecar-proxy")
// Send initial cluster discover (OK)
envoy.SendReq(t, ClusterType, 0, 0)
{
err, ok := getError()
require.NoError(t, err)
require.False(t, ok)
}
// Check no response sent yet
assertChanBlocked(t, envoy.stream.sendCh)
{
err, ok := getError()
require.NoError(t, err)
require.False(t, ok)
}
// Deliver a new snapshot
snap := proxycfg.TestConfigSnapshot(t)
mgr.DeliverConfig(t, "web-sidecar-proxy", snap)
assertResponseSent(t, envoy.stream.sendCh, expectClustersJSON(t, snap, token, 1, 1))
// Now nuke the ACL token.
validToken.Store("")
// It also (in parallel) issues the next cluster request (which acts as an ACK
// of the version we sent)
envoy.SendReq(t, ClusterType, 1, 1)
select {
case err := <-errCh:
require.Error(t, err)
gerr, ok := status.FromError(err)
require.Truef(t, ok, "not a grpc status error: type='%T' value=%v", err, err)
require.Equal(t, codes.Unauthenticated, gerr.Code())
require.Equal(t, "unauthenticated: ACL not found", gerr.Message())
mgr.AssertWatchCancelled(t, "web-sidecar-proxy")
case <-time.After(50 * time.Millisecond):
t.Fatalf("timed out waiting for handler to finish")
}
}
func TestServer_StreamAggregatedResources_ACLTokenDeleted_StreamTerminatedInBackground(t *testing.T) {
aclRules := `service "web" { policy = "write" }`
token := "service-write-on-web"
policy, err := acl.NewPolicyFromSource("", 0, aclRules, acl.SyntaxLegacy, nil)
require.NoError(t, err)
var validToken atomic.Value
validToken.Store(token)
logger := log.New(os.Stderr, "", log.LstdFlags)
mgr := newTestManager(t)
aclResolve := func(id string) (acl.Authorizer, error) {
if token := validToken.Load(); token == nil || id != token.(string) {
return nil, acl.ErrNotFound
}
return acl.NewPolicyAuthorizer(acl.RootAuthorizer("deny"), []*acl.Policy{policy}, nil)
}
envoy := NewTestEnvoy(t, "web-sidecar-proxy", token)
defer envoy.Close()
s := Server{
Logger: logger,
CfgMgr: mgr,
Authz: mgr,
ResolveToken: aclResolve,
AuthCheckFrequency: 100 * time.Millisecond, // Make this short.
}
s.Initialize()
errCh := make(chan error, 1)
go func() {
errCh <- s.StreamAggregatedResources(envoy.stream)
}()
getError := func() (gotErr error, ok bool) {
select {
case err := <-errCh:
return err, true
default:
return nil, false
}
}
// Register the proxy to create state needed to Watch() on
mgr.RegisterProxy(t, "web-sidecar-proxy")
// Send initial cluster discover (OK)
envoy.SendReq(t, ClusterType, 0, 0)
{
err, ok := getError()
require.NoError(t, err)
require.False(t, ok)
}
// Check no response sent yet
assertChanBlocked(t, envoy.stream.sendCh)
{
err, ok := getError()
require.NoError(t, err)
require.False(t, ok)
}
// Deliver a new snapshot
snap := proxycfg.TestConfigSnapshot(t)
mgr.DeliverConfig(t, "web-sidecar-proxy", snap)
assertResponseSent(t, envoy.stream.sendCh, expectClustersJSON(t, snap, token, 1, 1))
// It also (in parallel) issues the next cluster request (which acts as an ACK
// of the version we sent)
envoy.SendReq(t, ClusterType, 1, 1)
// Check no response sent yet
assertChanBlocked(t, envoy.stream.sendCh)
{
err, ok := getError()
require.NoError(t, err)
require.False(t, ok)
}
// Now nuke the ACL token while there's no activity.
validToken.Store("")
select {
case err := <-errCh:
require.Error(t, err)
gerr, ok := status.FromError(err)
require.Truef(t, ok, "not a grpc status error: type='%T' value=%v", err, err)
require.Equal(t, codes.Unauthenticated, gerr.Code())
require.Equal(t, "unauthenticated: ACL not found", gerr.Message())
mgr.AssertWatchCancelled(t, "web-sidecar-proxy")
case <-time.After(200 * time.Millisecond):
t.Fatalf("timed out waiting for handler to finish")
}
}
// This tests the ext_authz service method that implements connect authz.
func TestServer_Check(t *testing.T) {
tests := []struct {
name string
source string
dest string
sourcePrincipal string
destPrincipal string
authzResult connectAuthzResult
wantErr bool
wantErrCode codes.Code
wantDenied bool
wantReason string
}{
{
name: "auth allowed",
source: "web",
dest: "db",
authzResult: connectAuthzResult{true, "default allow", nil, nil},
wantDenied: false,
wantReason: "default allow",
},
{
name: "auth denied",
source: "web",
dest: "db",
authzResult: connectAuthzResult{false, "default deny", nil, nil},
wantDenied: true,
wantReason: "default deny",
},
{
name: "no source",
sourcePrincipal: "",
dest: "db",
// Should never make it to authz call.
wantErr: true,
wantErrCode: codes.InvalidArgument,
},
{
name: "no dest",
source: "web",
dest: "",
// Should never make it to authz call.
wantErr: true,
wantErrCode: codes.InvalidArgument,
},
{
name: "dest invalid format",
source: "web",
destPrincipal: "not-a-spiffe-id",
// Should never make it to authz call.
wantDenied: true,
wantReason: "Destination Principal is not a valid Connect identity",
},
{
name: "dest not a service URI",
source: "web",
destPrincipal: "spiffe://trust-domain.consul",
// Should never make it to authz call.
wantDenied: true,
wantReason: "Destination Principal is not a valid Service identity",
},
{
name: "ACL not got permission for authz call",
source: "web",
dest: "db",
authzResult: connectAuthzResult{false, "", nil, acl.ErrPermissionDenied},
wantErr: true,
wantErrCode: codes.PermissionDenied,
},
{
name: "Random error running authz",
source: "web",
dest: "db",
authzResult: connectAuthzResult{false, "", nil, errors.New("gremlin attack")},
wantErr: true,
wantErrCode: codes.Internal,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
token := "my-real-acl-token"
logger := log.New(os.Stderr, "", log.LstdFlags)
mgr := newTestManager(t)
// Setup expected auth result against that token no lock as no other
// goroutine is touching this yet.
mgr.authz[token] = tt.authzResult
aclResolve := func(id string) (acl.Authorizer, error) {
return nil, nil
}
envoy := NewTestEnvoy(t, "web-sidecar-proxy", token)
defer envoy.Close()
s := Server{
Logger: logger,
CfgMgr: mgr,
Authz: mgr,
ResolveToken: aclResolve,
}
s.Initialize()
// Create a context with the correct token
ctx := metadata.NewIncomingContext(context.Background(),
metadata.Pairs("x-consul-token", token))
r := TestCheckRequest(t, tt.source, tt.dest)
// If sourcePrincipal is set override, or if source is also not set
// explicitly override to empty.
if tt.sourcePrincipal != "" || tt.source == "" {
r.Attributes.Source.Principal = tt.sourcePrincipal
}
if tt.destPrincipal != "" || tt.dest == "" {
r.Attributes.Destination.Principal = tt.destPrincipal
}
resp, err := s.Check(ctx, r)
// Denied is not an error
if tt.wantErr {
require.Error(t, err)
grpcStatus := status.Convert(err)
require.Equal(t, tt.wantErrCode, grpcStatus.Code())
require.Nil(t, resp)
return
}
require.NoError(t, err)
if tt.wantDenied {
require.Equal(t, int32(codes.PermissionDenied), resp.Status.Code)
} else {
require.Equal(t, int32(codes.OK), resp.Status.Code)
}
require.Contains(t, resp.Status.Message, tt.wantReason)
})
}
}
func TestServer_ConfigOverrides(t *testing.T) {
tests := []struct {
name string
setup func(snap *proxycfg.ConfigSnapshot) string
}{
{
name: "sanity check no custom",
setup: func(snap *proxycfg.ConfigSnapshot) string {
// Default snap and expectation
return expectListenerJSON(t, snap, "my-token", 1, 1)
},
},
{
name: "custom public_listener no type",
setup: func(snap *proxycfg.ConfigSnapshot) string {
snap.Proxy.Config["envoy_public_listener_json"] =
customListenerJSON(t, customListenerJSONOptions{
Name: "custom-public-listen",
IncludeType: false,
})
resources := expectListenerJSONResources(t, snap, "my-token", 1, 1)
// Replace the public listener with the custom one WITH type since
// that's how it comes out the other end, and with TLS and authz
// overridden.
resources["public_listener"] = customListenerJSON(t, customListenerJSONOptions{
Name: "custom-public-listen",
// We should add type, TLS and authz
IncludeType: true,
OverrideAuthz: true,
TLSContext: expectedPublicTLSContextJSON(t, snap),
})
return expectListenerJSONFromResources(t, snap, "my-token", 1, 1, resources)
},
},
{
name: "custom public_listener with type",
setup: func(snap *proxycfg.ConfigSnapshot) string {
snap.Proxy.Config["envoy_public_listener_json"] =
customListenerJSON(t, customListenerJSONOptions{
Name: "custom-public-listen",
IncludeType: true,
})
resources := expectListenerJSONResources(t, snap, "my-token", 1, 1)
// Replace the public listener with the custom one WITH type since
// that's how it comes out the other end, and with TLS and authz
// overridden.
resources["public_listener"] = customListenerJSON(t, customListenerJSONOptions{
Name: "custom-public-listen",
// We should add type, TLS and authz
IncludeType: true,
OverrideAuthz: true,
TLSContext: expectedPublicTLSContextJSON(t, snap),
})
return expectListenerJSONFromResources(t, snap, "my-token", 1, 1, resources)
},
},
{
name: "custom public_listener with TLS should be overridden",
setup: func(snap *proxycfg.ConfigSnapshot) string {
snap.Proxy.Config["envoy_public_listener_json"] =
customListenerJSON(t, customListenerJSONOptions{
Name: "custom-public-listen",
IncludeType: true,
TLSContext: `{"requireClientCertificate": false}`,
})
resources := expectListenerJSONResources(t, snap, "my-token", 1, 1)
// Replace the public listener with the custom one WITH type since
// that's how it comes out the other end, and with TLS and authz
// overridden.
resources["public_listener"] = customListenerJSON(t, customListenerJSONOptions{
Name: "custom-public-listen",
// We should add type, TLS and authz
IncludeType: true,
OverrideAuthz: true,
TLSContext: expectedPublicTLSContextJSON(t, snap),
})
return expectListenerJSONFromResources(t, snap, "my-token", 1, 1, resources)
},
},
{
name: "custom upstream no type",
setup: func(snap *proxycfg.ConfigSnapshot) string {
snap.Proxy.Upstreams[0].Config["envoy_listener_json"] =
customListenerJSON(t, customListenerJSONOptions{
Name: "custom-upstream",
IncludeType: false,
})
resources := expectListenerJSONResources(t, snap, "my-token", 1, 1)
// Replace an upstream listener with the custom one WITH type since
// that's how it comes out the other end. Note we do override TLS
resources["service:db"] =
customListenerJSON(t, customListenerJSONOptions{
Name: "custom-upstream",
// We should add type
IncludeType: true,
})
return expectListenerJSONFromResources(t, snap, "my-token", 1, 1, resources)
},
},
{
name: "custom upstream with type",
setup: func(snap *proxycfg.ConfigSnapshot) string {
snap.Proxy.Upstreams[0].Config["envoy_listener_json"] =
customListenerJSON(t, customListenerJSONOptions{
Name: "custom-upstream",
IncludeType: true,
})
resources := expectListenerJSONResources(t, snap, "my-token", 1, 1)
// Replace an upstream listener with the custom one WITH type since
// that's how it comes out the other end.
resources["service:db"] =
customListenerJSON(t, customListenerJSONOptions{
Name: "custom-upstream",
// We should add type
IncludeType: true,
})
return expectListenerJSONFromResources(t, snap, "my-token", 1, 1, resources)
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
require := require.New(t)
// Sanity check default with no overrides first
snap := proxycfg.TestConfigSnapshot(t)
expect := tt.setup(snap)
listeners, err := listenersFromSnapshot(snap, "my-token")
require.NoError(err)
r, err := createResponse(ListenerType, "00000001", "00000001", listeners)
require.NoError(err)
assertResponse(t, r, expect)
})
}
}
type customListenerJSONOptions struct {
Name string
IncludeType bool
OverrideAuthz bool
TLSContext string
}
const customListenerJSONTpl = `{
{{ if .IncludeType -}}
"@type": "type.googleapis.com/envoy.api.v2.Listener",
{{- end }}
"name": "{{ .Name }}",
"address": {
"socketAddress": {
"address": "11.11.11.11",
"portValue": 11111
}
},
"filterChains": [
{
{{ if .TLSContext -}}
"tlsContext": {{ .TLSContext }},
{{- end }}
"filters": [
{{ if .OverrideAuthz -}}
{
"name": "envoy.ext_authz",
"config": {
"grpc_service": {
"envoy_grpc": {
"cluster_name": "local_agent"
},
"initial_metadata": [
{
"key": "x-consul-token",
"value": "my-token"
}
]
},
"stat_prefix": "connect_authz"
}
},
{{- end }}
{
"name": "envoy.tcp_proxy",
"config": {
"cluster": "random-cluster",
"stat_prefix": "foo-stats"
}
}
]
}
]
}`
var customListenerJSONTemplate = template.Must(template.New("").Parse(customListenerJSONTpl))
func customListenerJSON(t *testing.T, opts customListenerJSONOptions) string {
t.Helper()
var buf bytes.Buffer
err := customListenerJSONTemplate.Execute(&buf, opts)
require.NoError(t, err)
return buf.String()
}
Reference in New Issue View Git Blame Copy Permalink