Add internal RPC endpoint to compute upstreams from intentions

agent/connect/authz.go

@@ -0,0 +1,49 @@
+package connect
+
+import (
+	"github.com/hashicorp/consul/agent/structs"
+)
+
+// AuthorizeIntentionTarget determines whether the destination is covered by the given intention
+// and whether the intention action allows a connection.
+// This is a generalized version of the old CertURI.Authorize(), and can be evaluated against sources or destinations.
+//
+// The return value of `auth` is only valid if the second value `match` is true.
+// If `match` is false, then the intention doesn't match this target and any result should be ignored.
+func AuthorizeIntentionTarget(
+	target, targetNS string,
+	ixn *structs.Intention,
+	matchType structs.IntentionMatchType,
+) (auth bool, match bool) {
+
+	switch matchType {
+	case structs.IntentionMatchDestination:
+		if ixn.DestinationNS != structs.WildcardSpecifier && ixn.DestinationNS != targetNS {
+			// Non-matching namespace
+			return false, false
+		}
+
+		if ixn.DestinationName != structs.WildcardSpecifier && ixn.DestinationName != target {
+			// Non-matching name
+			return false, false
+		}
+
+	case structs.IntentionMatchSource:
+		if ixn.SourceNS != structs.WildcardSpecifier && ixn.SourceNS != targetNS {
+			// Non-matching namespace
+			return false, false
+		}
+
+		if ixn.SourceName != structs.WildcardSpecifier && ixn.SourceName != target {
+			// Non-matching name
+			return false, false
+		}
+
+	default:
+		// Reject on any un-recognized match type
+		return false, false
+	}
+
+	// The name and namespace match, so the destination is covered
+	return ixn.Action == structs.IntentionActionAllow, true
+}

agent/connect/authz_test.go

@@ -0,0 +1,196 @@
+package connect
+
+import (
+	"github.com/hashicorp/consul/agent/structs"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestAuthorizeIntentionTarget(t *testing.T) {
+	cases := []struct {
+		name      string
+		target    string
+		targetNS  string
+		ixn       *structs.Intention
+		matchType structs.IntentionMatchType
+		auth      bool
+		match     bool
+	}{
+		// Source match type
+		{
+			name:     "match exact source, not matching namespace",
+			target:   "web",
+			targetNS: structs.IntentionDefaultNamespace,
+			ixn: &structs.Intention{
+				SourceName: "db",
+				SourceNS:   "different",
+			},
+			matchType: structs.IntentionMatchSource,
+			auth:      false,
+			match:     false,
+		},
+		{
+			name:     "match exact source, not matching name",
+			target:   "web",
+			targetNS: structs.IntentionDefaultNamespace,
+			ixn: &structs.Intention{
+				SourceName: "db",
+				SourceNS:   structs.IntentionDefaultNamespace,
+			},
+			matchType: structs.IntentionMatchSource,
+			auth:      false,
+			match:     false,
+		},
+		{
+			name:     "match exact source, allow",
+			target:   "web",
+			targetNS: structs.IntentionDefaultNamespace,
+			ixn: &structs.Intention{
+				SourceName: "web",
+				SourceNS:   structs.IntentionDefaultNamespace,
+				Action:     structs.IntentionActionAllow,
+			},
+			matchType: structs.IntentionMatchSource,
+			auth:      true,
+			match:     true,
+		},
+		{
+			name:     "match exact source, deny",
+			target:   "web",
+			targetNS: structs.IntentionDefaultNamespace,
+			ixn: &structs.Intention{
+				SourceName: "web",
+				SourceNS:   structs.IntentionDefaultNamespace,
+				Action:     structs.IntentionActionDeny,
+			},
+			matchType: structs.IntentionMatchSource,
+			auth:      false,
+			match:     true,
+		},
+		{
+			name:     "match exact sourceNS for wildcard service, deny",
+			target:   "web",
+			targetNS: structs.IntentionDefaultNamespace,
+			ixn: &structs.Intention{
+				SourceName: structs.WildcardSpecifier,
+				SourceNS:   structs.IntentionDefaultNamespace,
+				Action:     structs.IntentionActionDeny,
+			},
+			matchType: structs.IntentionMatchSource,
+			auth:      false,
+			match:     true,
+		},
+		{
+			name:     "match exact sourceNS for wildcard service, allow",
+			target:   "web",
+			targetNS: structs.IntentionDefaultNamespace,
+			ixn: &structs.Intention{
+				SourceName: structs.WildcardSpecifier,
+				SourceNS:   structs.IntentionDefaultNamespace,
+				Action:     structs.IntentionActionAllow,
+			},
+			matchType: structs.IntentionMatchSource,
+			auth:      true,
+			match:     true,
+		},
+
+		// Destination match type
+		{
+			name:     "match exact destination, not matching namespace",
+			target:   "web",
+			targetNS: structs.IntentionDefaultNamespace,
+			ixn: &structs.Intention{
+				DestinationName: "db",
+				DestinationNS:   "different",
+			},
+			matchType: structs.IntentionMatchDestination,
+			auth:      false,
+			match:     false,
+		},
+		{
+			name:     "match exact destination, not matching name",
+			target:   "web",
+			targetNS: structs.IntentionDefaultNamespace,
+			ixn: &structs.Intention{
+				DestinationName: "db",
+				DestinationNS:   structs.IntentionDefaultNamespace,
+			},
+			matchType: structs.IntentionMatchDestination,
+			auth:      false,
+			match:     false,
+		},
+		{
+			name:     "match exact destination, allow",
+			target:   "web",
+			targetNS: structs.IntentionDefaultNamespace,
+			ixn: &structs.Intention{
+				DestinationName: "web",
+				DestinationNS:   structs.IntentionDefaultNamespace,
+				Action:          structs.IntentionActionAllow,
+			},
+			matchType: structs.IntentionMatchDestination,
+			auth:      true,
+			match:     true,
+		},
+		{
+			name:     "match exact destination, deny",
+			target:   "web",
+			targetNS: structs.IntentionDefaultNamespace,
+			ixn: &structs.Intention{
+				DestinationName: "web",
+				DestinationNS:   structs.IntentionDefaultNamespace,
+				Action:          structs.IntentionActionDeny,
+			},
+			matchType: structs.IntentionMatchDestination,
+			auth:      false,
+			match:     true,
+		},
+		{
+			name:     "match exact destinationNS for wildcard service, deny",
+			target:   "web",
+			targetNS: structs.IntentionDefaultNamespace,
+			ixn: &structs.Intention{
+				DestinationName: structs.WildcardSpecifier,
+				DestinationNS:   structs.IntentionDefaultNamespace,
+				Action:          structs.IntentionActionDeny,
+			},
+			matchType: structs.IntentionMatchDestination,
+			auth:      false,
+			match:     true,
+		},
+		{
+			name:     "match exact destinationNS for wildcard service, allow",
+			target:   "web",
+			targetNS: structs.IntentionDefaultNamespace,
+			ixn: &structs.Intention{
+				DestinationName: structs.WildcardSpecifier,
+				DestinationNS:   structs.IntentionDefaultNamespace,
+				Action:          structs.IntentionActionAllow,
+			},
+			matchType: structs.IntentionMatchDestination,
+			auth:      true,
+			match:     true,
+		},
+		{
+			name:     "unknown match type",
+			target:   "web",
+			targetNS: structs.IntentionDefaultNamespace,
+			ixn: &structs.Intention{
+				DestinationName: structs.WildcardSpecifier,
+				DestinationNS:   structs.IntentionDefaultNamespace,
+				Action:          structs.IntentionActionAllow,
+			},
+			matchType: structs.IntentionMatchType("unknown"),
+			auth:      false,
+			match:     false,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			auth, match := AuthorizeIntentionTarget(tc.target, tc.targetNS, tc.ixn, tc.matchType)
+			assert.Equal(t, tc.auth, auth)
+			assert.Equal(t, tc.match, match)
+		})
+	}
+}

agent/connect/uri.go

@@ -5,8 +5,6 @@ import (
 	"net/url"
 	"regexp"
 	"strings"
-
-	"github.com/hashicorp/consul/agent/structs"
 )
 
 // CertURI represents a Connect-valid URI value for a TLS certificate.
@@ -17,13 +15,6 @@ import (
 // However, we anticipate that we may accept URIs that are also not SPIFFE
 // compliant and therefore the interface is named as such.
 type CertURI interface {
-	// Authorize tests the authorization for this URI as a client
-	// for the given intention. The return value `auth` is only valid if
-	// the second value `match` is true. If the second value `match` is
-	// false, then the intention doesn't match this client and any
-	// result should be ignored.
-	Authorize(*structs.Intention) (auth bool, match bool)
-
 	// URI is the valid URI value used in the cert.
 	URI() *url.URL
 }

agent/connect/uri_agent.go

@@ -3,8 +3,6 @@ package connect
 import (
 	"fmt"
 	"net/url"
-
-	"github.com/hashicorp/consul/agent/structs"
 )
 
 // SpiffeIDService is the structure to represent the SPIFFE ID for an agent.
@@ -23,11 +21,6 @@ func (id *SpiffeIDAgent) URI() *url.URL {
 	return &result
 }
 
-// CertURI impl.
-func (id *SpiffeIDAgent) Authorize(_ *structs.Intention) (bool, bool) {
-	return false, false
-}
-
 func (id *SpiffeIDAgent) CommonName() string {
 	return AgentCN(id.Agent, id.Host)
 }

agent/connect/uri_agent_test.go

@@ -15,14 +15,3 @@ func TestSpiffeIDAgentURI(t *testing.T) {
 
 	require.Equal(t, "spiffe://1234.consul/agent/client/dc/dc1/id/123", agent.URI().String())
 }
-
-func TestSpiffeIDAgentAuthorize(t *testing.T) {
-	agent := &SpiffeIDAgent{
-		Host:  "1234.consul",
-		Agent: "uuid",
-	}
-
-	auth, match := agent.Authorize(nil)
-	require.False(t, auth)
-	require.False(t, match)
-}

agent/connect/uri_service.go

@@ -3,8 +3,6 @@ package connect
 import (
 	"fmt"
 	"net/url"
-
-	"github.com/hashicorp/consul/agent/structs"
 )
 
 // SpiffeIDService is the structure to represent the SPIFFE ID for a service.
@@ -25,22 +23,6 @@ func (id *SpiffeIDService) URI() *url.URL {
 	return &result
 }
 
-// CertURI impl.
-func (id *SpiffeIDService) Authorize(ixn *structs.Intention) (bool, bool) {
-	if ixn.SourceNS != structs.WildcardSpecifier && ixn.SourceNS != id.Namespace {
-		// Non-matching namespace
-		return false, false
-	}
-
-	if ixn.SourceName != structs.WildcardSpecifier && ixn.SourceName != id.Service {
-		// Non-matching name
-		return false, false
-	}
-
-	// Match, return allow value
-	return ixn.Action == structs.IntentionActionAllow, true
-}
-
 func (id *SpiffeIDService) CommonName() string {
 	return ServiceCN(id.Service, id.Namespace, id.Host)
 }

agent/connect/uri_service_test.go

@@ -1,104 +0,0 @@
-package connect
-
-import (
-	"testing"
-
-	"github.com/hashicorp/consul/agent/structs"
-	"github.com/stretchr/testify/assert"
-)
-
-func TestSpiffeIDServiceAuthorize(t *testing.T) {
-	ns := structs.IntentionDefaultNamespace
-	serviceWeb := &SpiffeIDService{
-		Host:       "1234.consul",
-		Namespace:  structs.IntentionDefaultNamespace,
-		Datacenter: "dc01",
-		Service:    "web",
-	}
-
-	cases := []struct {
-		Name  string
-		URI   *SpiffeIDService
-		Ixn   *structs.Intention
-		Auth  bool
-		Match bool
-	}{
-		{
-			"exact source, not matching namespace",
-			serviceWeb,
-			&structs.Intention{
-				SourceNS:   "different",
-				SourceName: "db",
-			},
-			false,
-			false,
-		},
-
-		{
-			"exact source, not matching name",
-			serviceWeb,
-			&structs.Intention{
-				SourceNS:   ns,
-				SourceName: "db",
-			},
-			false,
-			false,
-		},
-
-		{
-			"exact source, allow",
-			serviceWeb,
-			&structs.Intention{
-				SourceNS:   serviceWeb.Namespace,
-				SourceName: serviceWeb.Service,
-				Action:     structs.IntentionActionAllow,
-			},
-			true,
-			true,
-		},
-
-		{
-			"exact source, deny",
-			serviceWeb,
-			&structs.Intention{
-				SourceNS:   serviceWeb.Namespace,
-				SourceName: serviceWeb.Service,
-				Action:     structs.IntentionActionDeny,
-			},
-			false,
-			true,
-		},
-
-		{
-			"exact namespace, wildcard service, deny",
-			serviceWeb,
-			&structs.Intention{
-				SourceNS:   serviceWeb.Namespace,
-				SourceName: structs.WildcardSpecifier,
-				Action:     structs.IntentionActionDeny,
-			},
-			false,
-			true,
-		},
-
-		{
-			"exact namespace, wildcard service, allow",
-			serviceWeb,
-			&structs.Intention{
-				SourceNS:   serviceWeb.Namespace,
-				SourceName: structs.WildcardSpecifier,
-				Action:     structs.IntentionActionAllow,
-			},
-			true,
-			true,
-		},
-	}
-
-	for _, tc := range cases {
-		t.Run(tc.Name, func(t *testing.T) {
-			auth, match := tc.URI.Authorize(tc.Ixn)
-			assert.Equal(t, tc.Auth, auth)
-			assert.Equal(t, tc.Match, match)
-		})
-	}
-}

agent/connect/uri_signing.go

@@ -28,12 +28,6 @@ func (id *SpiffeIDSigning) Host() string {
 	return strings.ToLower(fmt.Sprintf("%s.%s", id.ClusterID, id.Domain))
 }
 
-// CertURI impl.
-func (id *SpiffeIDSigning) Authorize(ixn *structs.Intention) (bool, bool) {
-	// Never authorize as a client.
-	return false, true
-}
-
 // CanSign takes any CertURI and returns whether or not this signing entity is
 // allowed to sign CSRs for that entity (i.e. represents the trust domain for
 // that entity).

agent/connect/uri_signing_test.go

@@ -10,14 +10,6 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-// Signing ID should never authorize
-func TestSpiffeIDSigningAuthorize(t *testing.T) {
-	var id SpiffeIDSigning
-	auth, ok := id.Authorize(nil)
-	assert.False(t, auth)
-	assert.True(t, ok)
-}
-
 func TestSpiffeIDSigningForCluster(t *testing.T) {
 	// For now it should just append .consul to the ID.
 	config := &structs.CAConfiguration{
@@ -31,10 +23,6 @@ func TestSpiffeIDSigningForCluster(t *testing.T) {
 // about
 type fakeCertURI string
 
-func (f fakeCertURI) Authorize(*structs.Intention) (auth bool, match bool) {
-	return false, false
-}
-
 func (f fakeCertURI) URI() *url.URL {
 	u, _ := url.Parse(string(f))
 	return u

agent/connect_auth.go

@@ -3,7 +3,6 @@ package agent
 import (
 	"context"
 	"fmt"
-
 	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/agent/cache"
 	cachetype "github.com/hashicorp/consul/agent/cache-types"
@@ -105,7 +104,8 @@ func (a *Agent) ConnectAuthorize(token string,
 	// Figure out which source matches this request.
 	var ixnMatch *structs.Intention
 	for _, ixn := range reply.Matches[0] {
-		if _, ok := uriService.Authorize(ixn); ok {
+		// We match on the intention source because the uriService is the source of the connection to authorize.
+		if _, ok := connect.AuthorizeIntentionTarget(uriService.Service, uriService.Namespace, ixn, structs.IntentionMatchSource); ok {
 			ixnMatch = ixn
 			break
 		}

agent/consul/catalog_endpoint.go

@@ -397,7 +397,7 @@ func (c *Catalog) ServiceList(args *structs.DCSpecificRequest, reply *structs.In
 		&args.QueryOptions,
 		&reply.QueryMeta,
 		func(ws memdb.WatchSet, state *state.Store) error {
-			index, services, err := state.ServiceList(ws, &args.EnterpriseMeta)
+			index, services, err := state.ServiceList(ws, nil, &args.EnterpriseMeta)
 			if err != nil {
 				return err
 			}

agent/consul/helper_test.go

@@ -910,3 +910,138 @@ func registerTestTopologyEntries(t *testing.T, codec rpc.ClientCodec, token stri
 		require.NoError(t, msgpackrpc.CallWithCodec(codec, "ConfigEntry.Apply", &req, &out))
 	}
 }
+
+func registerIntentionUpstreamEntries(t *testing.T, codec rpc.ClientCodec, token string) {
+	t.Helper()
+
+	// api and api-proxy on node foo
+	// web and web-proxy on node foo
+	// redis and redis-proxy on node foo
+	// * -> * (deny) intention
+	// web -> api (allow)
+	registrations := map[string]*structs.RegisterRequest{
+		"Node foo": {
+			Datacenter:   "dc1",
+			Node:         "foo",
+			ID:           types.NodeID("e0155642-135d-4739-9853-a1ee6c9f945b"),
+			Address:      "127.0.0.2",
+			WriteRequest: structs.WriteRequest{Token: token},
+		},
+		"Service api on foo": {
+			Datacenter:     "dc1",
+			Node:           "foo",
+			SkipNodeUpdate: true,
+			Service: &structs.NodeService{
+				Kind:    structs.ServiceKindTypical,
+				ID:      "api",
+				Service: "api",
+			},
+			WriteRequest: structs.WriteRequest{Token: token},
+		},
+		"Service api-proxy": {
+			Datacenter:     "dc1",
+			Node:           "foo",
+			SkipNodeUpdate: true,
+			Service: &structs.NodeService{
+				Kind:    structs.ServiceKindConnectProxy,
+				ID:      "api-proxy",
+				Service: "api-proxy",
+				Port:    8443,
+				Address: "198.18.1.2",
+				Proxy: structs.ConnectProxyConfig{
+					DestinationServiceName: "api",
+				},
+			},
+			WriteRequest: structs.WriteRequest{Token: token},
+		},
+		"Service web on foo": {
+			Datacenter:     "dc1",
+			Node:           "foo",
+			SkipNodeUpdate: true,
+			Service: &structs.NodeService{
+				Kind:    structs.ServiceKindTypical,
+				ID:      "web",
+				Service: "web",
+			},
+			WriteRequest: structs.WriteRequest{Token: token},
+		},
+		"Service web-proxy on foo": {
+			Datacenter:     "dc1",
+			Node:           "foo",
+			SkipNodeUpdate: true,
+			Service: &structs.NodeService{
+				Kind:    structs.ServiceKindConnectProxy,
+				ID:      "web-proxy",
+				Service: "web-proxy",
+				Port:    8080,
+				Proxy: structs.ConnectProxyConfig{
+					DestinationServiceName: "web",
+				},
+			},
+			WriteRequest: structs.WriteRequest{Token: token},
+		},
+		"Service redis on foo": {
+			Datacenter:     "dc1",
+			Node:           "foo",
+			SkipNodeUpdate: true,
+			Service: &structs.NodeService{
+				Kind:    structs.ServiceKindTypical,
+				ID:      "redis",
+				Service: "redis",
+			},
+			WriteRequest: structs.WriteRequest{Token: token},
+		},
+		"Service redis-proxy on foo": {
+			Datacenter:     "dc1",
+			Node:           "foo",
+			SkipNodeUpdate: true,
+			Service: &structs.NodeService{
+				Kind:    structs.ServiceKindConnectProxy,
+				ID:      "redis-proxy",
+				Service: "redis-proxy",
+				Port:    1234,
+				Proxy: structs.ConnectProxyConfig{
+					DestinationServiceName: "redis",
+				},
+			},
+			WriteRequest: structs.WriteRequest{Token: token},
+		},
+	}
+	registerTestCatalogEntriesMap(t, codec, registrations)
+
+	// Add intentions: deny all and web -> api
+	entries := []structs.ConfigEntryRequest{
+		{
+			Datacenter: "dc1",
+			Entry: &structs.ServiceIntentionsConfigEntry{
+				Kind: structs.ServiceIntentions,
+				Name: "api",
+				Sources: []*structs.SourceIntention{
+					{
+						Name:   "web",
+						Action: structs.IntentionActionAllow,
+					},
+				},
+			},
+			WriteRequest: structs.WriteRequest{Token: token},
+		},
+		{
+			Datacenter: "dc1",
+			Entry: &structs.ServiceIntentionsConfigEntry{
+				Kind: structs.ServiceIntentions,
+				Name: "*",
+				Sources: []*structs.SourceIntention{
+					{
+						Name:   "*",
+						Action: structs.IntentionActionDeny,
+					},
+				},
+			},
+			WriteRequest: structs.WriteRequest{Token: token},
+		},
+	}
+	for _, req := range entries {
+		var out bool
+		require.NoError(t, msgpackrpc.CallWithCodec(codec, "ConfigEntry.Apply", &req, &out))
+	}
+}

agent/consul/intention_endpoint.go

@@ -8,7 +8,6 @@ import (
 	"github.com/armon/go-metrics"
 	"github.com/armon/go-metrics/prometheus"
 	"github.com/hashicorp/consul/acl"
-	"github.com/hashicorp/consul/agent/connect"
 	"github.com/hashicorp/consul/agent/consul/state"
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/lib"
@@ -684,16 +683,7 @@ func (s *Intention) Check(args *structs.IntentionQueryRequest, reply *structs.In
 		return fmt.Errorf("Invalid destination namespace %q: %v", query.DestinationNS, err)
 	}
 
-	// Build the URI
+	if query.SourceType != structs.IntentionSourceConsul {
-	var uri connect.CertURI
-	switch query.SourceType {
-	case structs.IntentionSourceConsul:
-		uri = &connect.SpiffeIDService{
-			Namespace: query.SourceNS,
-			Service:   query.SourceName,
-		}
-
-	default:
 		return fmt.Errorf("unsupported SourceType: %q", query.SourceType)
 	}
 
@@ -732,7 +722,17 @@ func (s *Intention) Check(args *structs.IntentionQueryRequest, reply *structs.In
 	}
 
 	state := s.srv.fsm.State()
-	decision, err := state.IntentionDecision(uri, query.DestinationName, query.DestinationNS, defaultDecision)
+
+	entry := structs.IntentionMatchEntry{
+		Namespace: query.SourceNS,
+		Name:      query.SourceName,
+	}
+	_, intentions, err := state.IntentionMatchOne(nil, entry, structs.IntentionMatchSource)
+	if err != nil {
+		return fmt.Errorf("failed to query intentions for %s/%s", query.SourceNS, query.SourceName)
+	}
+
+	decision, err := state.IntentionDecision(query.DestinationName, query.DestinationNS, intentions, structs.IntentionMatchDestination, defaultDecision, false)
 	if err != nil {
 		return fmt.Errorf("failed to get intention decision from (%s/%s) to (%s/%s): %v",
 			query.SourceNS, query.SourceName, query.DestinationNS, query.DestinationName, err)

agent/consul/internal_endpoint.go

@@ -188,6 +188,49 @@ func (m *Internal) ServiceTopology(args *structs.ServiceSpecificRequest, reply *
 		})
 }
 
+// IntentionUpstreams returns the upstreams of a service. Upstreams are inferred from intentions.
+// If intentions allow a connection from the target to some candidate service, the candidate service is considered
+// an upstream of the target.
+func (m *Internal) IntentionUpstreams(args *structs.ServiceSpecificRequest, reply *structs.IndexedServiceList) error {
+	// Exit early if Connect hasn't been enabled.
+	if !m.srv.config.ConnectEnabled {
+		return ErrConnectNotEnabled
+	}
+	if args.ServiceName == "" {
+		return fmt.Errorf("Must provide a service name")
+	}
+	if done, err := m.srv.ForwardRPC("Internal.IntentionUpstreams", args, args, reply); done {
+		return err
+	}
+
+	authz, err := m.srv.ResolveTokenAndDefaultMeta(args.Token, &args.EnterpriseMeta, nil)
+	if err != nil {
+		return err
+	}
+	if err := m.srv.validateEnterpriseRequest(&args.EnterpriseMeta, false); err != nil {
+		return err
+	}
+
+	return m.srv.blockingQuery(
+		&args.QueryOptions,
+		&reply.QueryMeta,
+		func(ws memdb.WatchSet, state *state.Store) error {
+			defaultDecision := acl.Allow
+			if authz != nil {
+				defaultDecision = authz.IntentionDefaultAllow(nil)
+			}
+
+			sn := structs.NewServiceName(args.ServiceName, &args.EnterpriseMeta)
+			index, services, err := state.IntentionTopology(ws, sn, false, defaultDecision)
+			if err != nil {
+				return err
+			}
+
+			reply.Index, reply.Services = index, services
+			return m.srv.filterACLWithAuthorizer(authz, reply)
+		})
+}
+
 // GatewayServiceNodes returns all the nodes for services associated with a gateway along with their gateway config
 func (m *Internal) GatewayServiceDump(args *structs.ServiceSpecificRequest, reply *structs.IndexedServiceDump) error {
 	if done, err := m.srv.ForwardRPC("Internal.GatewayServiceDump", args, args, reply); done {

agent/consul/internal_endpoint_test.go

@@ -1885,3 +1885,124 @@ service "web" { policy = "read" }
 		require.True(t, acl.IsErrPermissionDenied(err))
 	})
 }
+
+func TestInternal_IntentionUpstreams(t *testing.T) {
+	if testing.Short() {
+		t.Skip("too slow for testing.Short")
+	}
+
+	t.Parallel()
+	dir1, s1 := testServer(t)
+	defer os.RemoveAll(dir1)
+	defer s1.Shutdown()
+
+	testrpc.WaitForLeader(t, s1.RPC, "dc1")
+
+	codec := rpcClient(t, s1)
+	defer codec.Close()
+
+	// Services:
+	// api and api-proxy on node foo
+	// web and web-proxy on node foo
+	//
+	// Intentions
+	// * -> * (deny) intention
+	// web -> api (allow)
+	registerIntentionUpstreamEntries(t, codec, "")
+
+	t.Run("web", func(t *testing.T) {
+		retry.Run(t, func(r *retry.R) {
+			args := structs.ServiceSpecificRequest{
+				Datacenter:  "dc1",
+				ServiceName: "web",
+			}
+			var out structs.IndexedServiceList
+			require.NoError(r, msgpackrpc.CallWithCodec(codec, "Internal.IntentionUpstreams", &args, &out))
+
+			// foo/api
+			require.Len(r, out.Services, 1)
+
+			expectUp := structs.ServiceList{
+				structs.NewServiceName("api", structs.DefaultEnterpriseMeta()),
+			}
+			require.Equal(r, expectUp, out.Services)
+		})
+	})
+}
+
+func TestInternal_IntentionUpstreams_ACL(t *testing.T) {
+	if testing.Short() {
+		t.Skip("too slow for testing.Short")
+	}
+
+	t.Parallel()
+	dir1, s1 := testServerWithConfig(t, func(c *Config) {
+		c.ACLDatacenter = "dc1"
+		c.ACLsEnabled = true
+		c.ACLMasterToken = TestDefaultMasterToken
+		c.ACLDefaultPolicy = "deny"
+	})
+	defer os.RemoveAll(dir1)
+	defer s1.Shutdown()
+
+	testrpc.WaitForLeader(t, s1.RPC, "dc1")
+
+	codec := rpcClient(t, s1)
+	defer codec.Close()
+
+	// Services:
+	// api and api-proxy on node foo
+	// web and web-proxy on node foo
+	//
+	// Intentions
+	// * -> * (deny) intention
+	// web -> api (allow)
+	registerIntentionUpstreamEntries(t, codec, TestDefaultMasterToken)
+
+	t.Run("valid token", func(t *testing.T) {
+		// Token grants read to read api service
+		userToken, err := upsertTestTokenWithPolicyRules(codec, TestDefaultMasterToken, "dc1", `
+service_prefix "api" { policy = "read" }
+`)
+		require.NoError(t, err)
+
+		retry.Run(t, func(r *retry.R) {
+			args := structs.ServiceSpecificRequest{
+				Datacenter:   "dc1",
+				ServiceName:  "web",
+				QueryOptions: structs.QueryOptions{Token: userToken.SecretID},
+			}
+			var out structs.IndexedServiceList
+			require.NoError(r, msgpackrpc.CallWithCodec(codec, "Internal.IntentionUpstreams", &args, &out))
+
+			// foo/api
+			require.Len(r, out.Services, 1)
+
+			expectUp := structs.ServiceList{
+				structs.NewServiceName("api", structs.DefaultEnterpriseMeta()),
+			}
+			require.Equal(r, expectUp, out.Services)
+		})
+	})
+
+	t.Run("invalid token filters results", func(t *testing.T) {
+		// Token grants read to read an unrelated service, mongo
+		userToken, err := upsertTestTokenWithPolicyRules(codec, TestDefaultMasterToken, "dc1", `
+service_prefix "mongo" { policy = "read" }
+`)
+		require.NoError(t, err)
+
+		retry.Run(t, func(r *retry.R) {
+			args := structs.ServiceSpecificRequest{
+				Datacenter:   "dc1",
+				ServiceName:  "web",
+				QueryOptions: structs.QueryOptions{Token: userToken.SecretID},
+			}
+			var out structs.IndexedServiceList
+			require.NoError(r, msgpackrpc.CallWithCodec(codec, "Internal.IntentionUpstreams", &args, &out))
+
+			// Token can't read api service
+			require.Empty(r, out.Services)
+		})
+	})
+}

agent/consul/state/catalog.go

@@ -11,7 +11,6 @@ import (
 	"github.com/mitchellh/copystructure"
 
 	"github.com/hashicorp/consul/acl"
-	"github.com/hashicorp/consul/agent/connect"
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/consul/lib"
@@ -724,14 +723,16 @@ func (s *Store) Services(ws memdb.WatchSet, entMeta *structs.EnterpriseMeta) (ui
 	return idx, results, nil
 }
 
-func (s *Store) ServiceList(ws memdb.WatchSet, entMeta *structs.EnterpriseMeta) (uint64, structs.ServiceList, error) {
+func (s *Store) ServiceList(ws memdb.WatchSet,
+	include func(svc *structs.ServiceNode) bool, entMeta *structs.EnterpriseMeta) (uint64, structs.ServiceList, error) {
 	tx := s.db.Txn(false)
 	defer tx.Abort()
 
-	return serviceListTxn(tx, ws, entMeta)
+	return serviceListTxn(tx, ws, include, entMeta)
 }
 
-func serviceListTxn(tx ReadTxn, ws memdb.WatchSet, entMeta *structs.EnterpriseMeta) (uint64, structs.ServiceList, error) {
+func serviceListTxn(tx ReadTxn, ws memdb.WatchSet,
+	include func(svc *structs.ServiceNode) bool, entMeta *structs.EnterpriseMeta) (uint64, structs.ServiceList, error) {
 	idx := catalogServicesMaxIndex(tx, entMeta)
 
 	services, err := catalogServiceList(tx, entMeta, true)
@@ -743,7 +744,11 @@ func serviceListTxn(tx ReadTxn, ws memdb.WatchSet, entMeta *structs.EnterpriseMe
 	unique := make(map[structs.ServiceName]struct{})
 	for service := services.Next(); service != nil; service = services.Next() {
 		svc := service.(*structs.ServiceNode)
-		unique[svc.CompoundServiceName()] = struct{}{}
+		// TODO (freddy) This is a hack to exclude certain kinds.
+		//				 Need a new index to query by kind and namespace, have to coordinate with consul foundations first
+		if include == nil || include(svc) {
+			unique[svc.CompoundServiceName()] = struct{}{}
+		}
 	}
 
 	results := make(structs.ServiceList, 0, len(unique))
@@ -2848,16 +2853,20 @@ func (s *Store) ServiceTopology(
 
 	upstreamDecisions := make(map[string]structs.IntentionDecisionSummary)
 
-	// The given service is the source relative to upstreams
+	matchEntry := structs.IntentionMatchEntry{
-	sourceURI := connect.SpiffeIDService{
 		Namespace: entMeta.NamespaceOrDefault(),
-		Service:   service,
+		Name:      service,
+	}
+	// The given service is a source relative to its upstreams
+	_, srcIntentions, err := compatIntentionMatchOneTxn(tx, ws, matchEntry, structs.IntentionMatchSource)
+	if err != nil {
+		return 0, nil, fmt.Errorf("failed to query intentions for %s", sn.String())
 	}
 	for _, un := range upstreamNames {
-		decision, err := s.IntentionDecision(&sourceURI, un.Name, un.NamespaceOrDefault(), defaultAllow)
+		decision, err := s.IntentionDecision(un.Name, un.NamespaceOrDefault(), srcIntentions, structs.IntentionMatchDestination, defaultAllow, false)
 		if err != nil {
-			return 0, nil, fmt.Errorf("failed to get intention decision from (%s/%s) to (%s/%s): %v",
+			return 0, nil, fmt.Errorf("failed to get intention decision from (%s) to (%s): %v",
-				sourceURI.Namespace, sourceURI.Service, un.Name, un.NamespaceOrDefault(), err)
+				sn.String(), un.String(), err)
 		}
 		upstreamDecisions[un.String()] = decision
 	}
@@ -2877,17 +2886,17 @@ func (s *Store) ServiceTopology(
 		maxIdx = idx
 	}
 
+	// The given service is a destination relative to its downstreams
+	_, dstIntentions, err := compatIntentionMatchOneTxn(tx, ws, matchEntry, structs.IntentionMatchDestination)
+	if err != nil {
+		return 0, nil, fmt.Errorf("failed to query intentions for %s", sn.String())
+	}
 	downstreamDecisions := make(map[string]structs.IntentionDecisionSummary)
 	for _, dn := range downstreamNames {
-		// Downstreams are the source relative to the given service
+		decision, err := s.IntentionDecision(dn.Name, dn.NamespaceOrDefault(), dstIntentions, structs.IntentionMatchSource, defaultAllow, false)
-		sourceURI := connect.SpiffeIDService{
-			Namespace: dn.NamespaceOrDefault(),
-			Service:   dn.Name,
-		}
-		decision, err := s.IntentionDecision(&sourceURI, service, entMeta.NamespaceOrDefault(), defaultAllow)
 		if err != nil {
-			return 0, nil, fmt.Errorf("failed to get intention decision from (%s/%s) to (%s/%s): %v",
+			return 0, nil, fmt.Errorf("failed to get intention decision from (%s) to (%s): %v",
-				sourceURI.Namespace, sourceURI.Service, service, dn.NamespaceOrDefault(), err)
+				dn.String(), sn.String(), err)
 		}
 		downstreamDecisions[dn.String()] = decision
 	}

agent/consul/state/intention.go

@@ -3,12 +3,12 @@ package state
 import (
 	"errors"
 	"fmt"
+	"github.com/hashicorp/consul/agent/connect"
 	"sort"
 
 	"github.com/hashicorp/go-memdb"
 
 	"github.com/hashicorp/consul/acl"
-	"github.com/hashicorp/consul/agent/connect"
 	"github.com/hashicorp/consul/agent/structs"
 )
 
@@ -732,35 +732,19 @@ func (s *Store) LegacyIntentionDeleteAll(idx uint64) error {
 	return tx.Commit()
 }
 
-// IntentionDecision returns whether a connection should be allowed from a source URI to some destination
+// IntentionDecision returns whether a connection should be allowed to a source or destination given a set of intentions.
-// It returns true or false for the enforcement, and also a boolean for whether
+//
+// allowPermissions determines whether the presence of L7 permissions leads to a DENY decision.
+// This should be false when evaluating a connection between a source and destination, but not the request that will be sent.
 func (s *Store) IntentionDecision(
-	srcURI connect.CertURI, dstName, dstNS string, defaultDecision acl.EnforcementDecision,
+	target, targetNS string, intentions structs.Intentions, matchType structs.IntentionMatchType,
+	defaultDecision acl.EnforcementDecision, allowPermissions bool,
 ) (structs.IntentionDecisionSummary, error) {
 
-	_, matches, err := s.IntentionMatch(nil, &structs.IntentionQueryMatch{
-		Type: structs.IntentionMatchDestination,
-		Entries: []structs.IntentionMatchEntry{
-			{
-				Namespace: dstNS,
-				Name:      dstName,
-			},
-		},
-	})
-	if err != nil {
-		return structs.IntentionDecisionSummary{}, err
-	}
-	if len(matches) != 1 {
-		// This should never happen since the documented behavior of the
-		// Match call is that it'll always return exactly the number of results
-		// as entries passed in. But we guard against misbehavior.
-		return structs.IntentionDecisionSummary{}, errors.New("internal error loading matches")
-	}
-
 	// Figure out which source matches this request.
 	var ixnMatch *structs.Intention
-	for _, ixn := range matches[0] {
+	for _, ixn := range intentions {
-		if _, ok := srcURI.Authorize(ixn); ok {
+		if _, ok := connect.AuthorizeIntentionTarget(target, targetNS, ixn, matchType); ok {
 			ixnMatch = ixn
 			break
 		}
@@ -776,9 +760,9 @@ func (s *Store) IntentionDecision(
 	// Intention found, combine action + permissions
 	resp.Allowed = ixnMatch.Action == structs.IntentionActionAllow
 	if len(ixnMatch.Permissions) > 0 {
-		// If there are L7 permissions, DENY.
+		// If any permissions are present, fall back to allowPermissions.
-		// We are only evaluating source and destination, not the request that will be sent.
+		// We are not evaluating requests so we cannot know whether the L7 permission requirements will be met.
-		resp.Allowed = false
+		resp.Allowed = allowPermissions
 		resp.HasPermissions = true
 	}
 	resp.ExternalSource = ixnMatch.Meta[structs.MetaExternalSource]
@@ -853,6 +837,16 @@ func (s *Store) IntentionMatchOne(
 	tx := s.db.Txn(false)
 	defer tx.Abort()
 
+	return compatIntentionMatchOneTxn(tx, ws, entry, matchType)
+}
+
+func compatIntentionMatchOneTxn(
+	tx ReadTxn,
+	ws memdb.WatchSet,
+	entry structs.IntentionMatchEntry,
+	matchType structs.IntentionMatchType,
+) (uint64, structs.Intentions, error) {
+
 	usingConfigEntries, err := areIntentionsInConfigEntries(tx, ws)
 	if err != nil {
 		return 0, nil, err
@@ -936,3 +930,90 @@ func intentionMatchGetParams(entry structs.IntentionMatchEntry) ([][]interface{}
 	result = append(result, []interface{}{entry.Namespace, entry.Name})
 	return result, nil
 }
+
+// IntentionTopology returns the upstreams or downstreams of a service. Upstreams and downstreams are inferred from
+// intentions. If intentions allow a connection from the target to some candidate service, the candidate service is considered
+// an upstream of the target.
+func (s *Store) IntentionTopology(ws memdb.WatchSet,
+	target structs.ServiceName, downstreams bool, defaultDecision acl.EnforcementDecision) (uint64, structs.ServiceList, error) {
+	tx := s.db.ReadTxn()
+	defer tx.Abort()
+
+	var maxIdx uint64
+
+	// If querying the upstreams for a service, we first query intentions that apply to the target service as a source.
+	// That way we can check whether intentions from the source allow connections to upstream candidates.
+	// The reverse is true for downstreams.
+	intentionMatchType := structs.IntentionMatchSource
+	if downstreams {
+		intentionMatchType = structs.IntentionMatchDestination
+	}
+	entry := structs.IntentionMatchEntry{
+		Namespace: target.NamespaceOrDefault(),
+		Name:      target.Name,
+	}
+	index, intentions, err := compatIntentionMatchOneTxn(tx, ws, entry, intentionMatchType)
+	if err != nil {
+		return 0, nil, fmt.Errorf("failed to query intentions for %s", target.String())
+	}
+	if index > maxIdx {
+		maxIdx = index
+	}
+
+	// Check for a wildcard intention (* -> *) since it overrides the default decision from ACLs
+	if len(intentions) > 0 {
+		// Intentions with wildcard source and destination have the lowest precedence, so they are last in the list
+		ixn := intentions[len(intentions)-1]
+
+		// TODO (freddy) This needs an enterprise split to account for (*/* -> */*)
+		//				 Maybe ixn.HasWildcardSource() && ixn.HasWildcardDestination()
+		if ixn.SourceName == structs.WildcardSpecifier && ixn.DestinationName == structs.WildcardSpecifier {
+			defaultDecision = acl.Allow
+			if ixn.Action == structs.IntentionActionDeny {
+				defaultDecision = acl.Deny
+			}
+		}
+	}
+
+	index, allServices, err := serviceListTxn(tx, ws, func(svc *structs.ServiceNode) bool {
+		// Only include ingress gateways as downstreams, since they cannot receive service mesh traffic
+		// TODO(freddy): One remaining issue is that this includes non-Connect services (typical services without a proxy)
+		//				 Ideally those should be excluded as well, since they can't be upstreams/downstreams without a proxy.
+		//				 Maybe start tracking services represented by proxies? (both sidecar and ingress)
+		if svc.ServiceKind == structs.ServiceKindTypical || (svc.ServiceKind == structs.ServiceKindIngressGateway && downstreams) {
+			return true
+		}
+		return false
+	}, structs.WildcardEnterpriseMeta())
+	if err != nil {
+		return index, nil, fmt.Errorf("failed to fetch catalog service list: %v", err)
+	}
+	if index > maxIdx {
+		maxIdx = index
+	}
+
+	// When checking authorization to upstreams, the match type for the decision is `destination` because we are deciding
+	// if upstream candidates are covered by intentions that have the target service as a source.
+	// The reverse is true for downstreams.
+	decisionMatchType := structs.IntentionMatchDestination
+	if downstreams {
+		decisionMatchType = structs.IntentionMatchSource
+	}
+	result := make(structs.ServiceList, 0, len(allServices))
+	for _, candidate := range allServices {
+		decision, err := s.IntentionDecision(candidate.Name, candidate.NamespaceOrDefault(), intentions, decisionMatchType, defaultDecision, true)
+		if err != nil {
+			src, dst := target, candidate
+			if downstreams {
+				src, dst = candidate, target
+			}
+			return 0, nil, fmt.Errorf("failed to get intention decision from (%s) to (%s): %v",
+				src.String(), dst.String(), err)
+		}
+		if !decision.Allowed || target.Matches(candidate) {
+			continue
+		}
+		result = append(result, candidate)
+	}
+	return maxIdx, result, err
+}

agent/consul/state/intention_test.go

@@ -1,6 +1,7 @@
 package state
 
 import (
+	"sort"
 	"testing"
 	"time"
 
@@ -9,7 +10,6 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/hashicorp/consul/acl"
-	"github.com/hashicorp/consul/agent/connect"
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/sdk/testutil"
 )
@@ -1760,16 +1760,19 @@ func TestStore_IntentionDecision(t *testing.T) {
 	}
 
 	tt := []struct {
-		name            string
+		name             string
-		src             string
+		src              string
-		dst             string
+		dst              string
-		defaultDecision acl.EnforcementDecision
+		matchType        structs.IntentionMatchType
-		expect          structs.IntentionDecisionSummary
+		defaultDecision  acl.EnforcementDecision
+		allowPermissions bool
+		expect           structs.IntentionDecisionSummary
 	}{
 		{
 			name:            "no matching intention and default deny",
 			src:             "does-not-exist",
 			dst:             "ditto",
+			matchType:       structs.IntentionMatchDestination,
 			defaultDecision: acl.Deny,
 			expect:          structs.IntentionDecisionSummary{Allowed: false},
 		},
@@ -1777,13 +1780,15 @@ func TestStore_IntentionDecision(t *testing.T) {
 			name:            "no matching intention and default allow",
 			src:             "does-not-exist",
 			dst:             "ditto",
+			matchType:       structs.IntentionMatchDestination,
 			defaultDecision: acl.Allow,
 			expect:          structs.IntentionDecisionSummary{Allowed: true},
 		},
 		{
-			name: "denied with permissions",
+			name:      "denied with permissions",
-			src:  "web",
+			src:       "web",
-			dst:  "redis",
+			dst:       "redis",
+			matchType: structs.IntentionMatchDestination,
 			expect: structs.IntentionDecisionSummary{
 				Allowed:        false,
 				HasPermissions: true,
@@ -1791,9 +1796,22 @@ func TestStore_IntentionDecision(t *testing.T) {
 			},
 		},
 		{
-			name: "denied without permissions",
+			name:             "allowed with permissions",
-			src:  "api",
+			src:              "web",
-			dst:  "redis",
+			dst:              "redis",
+			allowPermissions: true,
+			matchType:        structs.IntentionMatchDestination,
+			expect: structs.IntentionDecisionSummary{
+				Allowed:        true,
+				HasPermissions: true,
+				HasExact:       true,
+			},
+		},
+		{
+			name:      "denied without permissions",
+			src:       "api",
+			dst:       "redis",
+			matchType: structs.IntentionMatchDestination,
 			expect: structs.IntentionDecisionSummary{
 				Allowed:        false,
 				HasPermissions: false,
@@ -1801,9 +1819,10 @@ func TestStore_IntentionDecision(t *testing.T) {
 			},
 		},
 		{
-			name: "allowed from external source",
+			name:      "allowed from external source",
-			src:  "api",
+			src:       "api",
-			dst:  "web",
+			dst:       "web",
+			matchType: structs.IntentionMatchDestination,
 			expect: structs.IntentionDecisionSummary{
 				Allowed:        true,
 				HasPermissions: false,
@@ -1812,9 +1831,21 @@ func TestStore_IntentionDecision(t *testing.T) {
 			},
 		},
 		{
-			name: "allowed by source wildcard not exact",
+			name:      "allowed by source wildcard not exact",
-			src:  "anything",
+			src:       "anything",
-			dst:  "mysql",
+			dst:       "mysql",
+			matchType: structs.IntentionMatchDestination,
+			expect: structs.IntentionDecisionSummary{
+				Allowed:        true,
+				HasPermissions: false,
+				HasExact:       false,
+			},
+		},
+		{
+			name:      "allowed by matching on source",
+			src:       "web",
+			dst:       "api",
+			matchType: structs.IntentionMatchSource,
 			expect: structs.IntentionDecisionSummary{
 				Allowed:        true,
 				HasPermissions: false,
@@ -1824,11 +1855,15 @@ func TestStore_IntentionDecision(t *testing.T) {
 	}
 	for _, tc := range tt {
 		t.Run(tc.name, func(t *testing.T) {
-			uri := connect.SpiffeIDService{
+			entry := structs.IntentionMatchEntry{
-				Service:   tc.src,
 				Namespace: structs.IntentionDefaultNamespace,
+				Name:      tc.src,
 			}
-			decision, err := s.IntentionDecision(&uri, tc.dst, structs.IntentionDefaultNamespace, tc.defaultDecision)
+			_, intentions, err := s.IntentionMatchOne(nil, entry, structs.IntentionMatchSource)
+			if err != nil {
+				require.NoError(t, err)
+			}
+			decision, err := s.IntentionDecision(tc.dst, structs.IntentionDefaultNamespace, intentions, tc.matchType, tc.defaultDecision, tc.allowPermissions)
 			require.NoError(t, err)
 			require.Equal(t, tc.expect, decision)
 		})
@@ -1847,3 +1882,374 @@ func testConfigStateStore(t *testing.T) *Store {
 	disableLegacyIntentions(s)
 	return s
 }
+
+func TestStore_IntentionTopology(t *testing.T) {
+	node := structs.Node{
+		Node:    "foo",
+		Address: "127.0.0.1",
+	}
+	services := []structs.NodeService{
+		{
+			ID:             "api-1",
+			Service:        "api",
+			EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+		},
+		{
+			ID:             "mysql-1",
+			Service:        "mysql",
+			EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+		},
+		{
+			ID:             "web-1",
+			Service:        "web",
+			EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+		},
+		{
+			Kind:           structs.ServiceKindConnectProxy,
+			ID:             "web-proxy-1",
+			Service:        "web-proxy",
+			EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+		},
+		{
+			Kind:           structs.ServiceKindTerminatingGateway,
+			ID:             "terminating-gateway-1",
+			Service:        "terminating-gateway",
+			EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+		},
+		{
+			Kind:           structs.ServiceKindIngressGateway,
+			ID:             "ingress-gateway-1",
+			Service:        "ingress-gateway",
+			EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+		},
+		{
+			Kind:           structs.ServiceKindMeshGateway,
+			ID:             "mesh-gateway-1",
+			Service:        "mesh-gateway",
+			EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+		},
+	}
+
+	type expect struct {
+		idx      uint64
+		services structs.ServiceList
+	}
+	tests := []struct {
+		name            string
+		defaultDecision acl.EnforcementDecision
+		intentions      []structs.ServiceIntentionsConfigEntry
+		target          structs.ServiceName
+		downstreams     bool
+		expect          expect
+	}{
+		{
+			name:            "(upstream) acl allow all but intentions deny one",
+			defaultDecision: acl.Allow,
+			intentions: []structs.ServiceIntentionsConfigEntry{
+				{
+					Kind: structs.ServiceIntentions,
+					Name: "api",
+					Sources: []*structs.SourceIntention{
+						{
+							Name:   "web",
+							Action: structs.IntentionActionDeny,
+						},
+					},
+				},
+			},
+			target:      structs.NewServiceName("web", nil),
+			downstreams: false,
+			expect: expect{
+				idx: 9,
+				services: structs.ServiceList{
+					{
+						Name:           "mysql",
+						EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+					},
+				},
+			},
+		},
+		{
+			name:            "(upstream) acl deny all intentions allow one",
+			defaultDecision: acl.Deny,
+			intentions: []structs.ServiceIntentionsConfigEntry{
+				{
+					Kind: structs.ServiceIntentions,
+					Name: "api",
+					Sources: []*structs.SourceIntention{
+						{
+							Name:   "web",
+							Action: structs.IntentionActionAllow,
+						},
+					},
+				},
+			},
+			target:      structs.NewServiceName("web", nil),
+			downstreams: false,
+			expect: expect{
+				idx: 9,
+				services: structs.ServiceList{
+					{
+						Name:           "api",
+						EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+					},
+				},
+			},
+		},
+		{
+			name:            "(downstream) acl allow all but intentions deny one",
+			defaultDecision: acl.Allow,
+			intentions: []structs.ServiceIntentionsConfigEntry{
+				{
+					Kind: structs.ServiceIntentions,
+					Name: "api",
+					Sources: []*structs.SourceIntention{
+						{
+							Name:   "web",
+							Action: structs.IntentionActionDeny,
+						},
+					},
+				},
+			},
+			target:      structs.NewServiceName("api", nil),
+			downstreams: true,
+			expect: expect{
+				idx: 9,
+				services: structs.ServiceList{
+					{
+						Name:           "ingress-gateway",
+						EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+					},
+					{
+						Name:           "mysql",
+						EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+					},
+				},
+			},
+		},
+		{
+			name:            "(downstream) acl deny all intentions allow one",
+			defaultDecision: acl.Deny,
+			intentions: []structs.ServiceIntentionsConfigEntry{
+				{
+					Kind: structs.ServiceIntentions,
+					Name: "api",
+					Sources: []*structs.SourceIntention{
+						{
+							Name:   "web",
+							Action: structs.IntentionActionAllow,
+						},
+					},
+				},
+			},
+			target:      structs.NewServiceName("api", nil),
+			downstreams: true,
+			expect: expect{
+				idx: 9,
+				services: structs.ServiceList{
+					{
+						Name:           "web",
+						EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+					},
+				},
+			},
+		},
+		{
+			name:            "acl deny but intention allow all overrides it",
+			defaultDecision: acl.Deny,
+			intentions: []structs.ServiceIntentionsConfigEntry{
+				{
+					Kind: structs.ServiceIntentions,
+					Name: "*",
+					Sources: []*structs.SourceIntention{
+						{
+							Name:   "*",
+							Action: structs.IntentionActionAllow,
+						},
+					},
+				},
+			},
+			target:      structs.NewServiceName("web", nil),
+			downstreams: false,
+			expect: expect{
+				idx: 9,
+				services: structs.ServiceList{
+					{
+						Name:           "api",
+						EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+					},
+					{
+						Name:           "mysql",
+						EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+					},
+				},
+			},
+		},
+		{
+			name:            "acl allow but intention deny all overrides it",
+			defaultDecision: acl.Allow,
+			intentions: []structs.ServiceIntentionsConfigEntry{
+				{
+					Kind: structs.ServiceIntentions,
+					Name: "*",
+					Sources: []*structs.SourceIntention{
+						{
+							Name:   "*",
+							Action: structs.IntentionActionDeny,
+						},
+					},
+				},
+			},
+			target:      structs.NewServiceName("web", nil),
+			downstreams: false,
+			expect: expect{
+				idx:      9,
+				services: structs.ServiceList{},
+			},
+		},
+		{
+			name:            "acl deny but intention allow all overrides it",
+			defaultDecision: acl.Deny,
+			intentions: []structs.ServiceIntentionsConfigEntry{
+				{
+					Kind: structs.ServiceIntentions,
+					Name: "*",
+					Sources: []*structs.SourceIntention{
+						{
+							Name:   "*",
+							Action: structs.IntentionActionAllow,
+						},
+					},
+				},
+			},
+			target:      structs.NewServiceName("web", nil),
+			downstreams: false,
+			expect: expect{
+				idx: 9,
+				services: structs.ServiceList{
+					{
+						Name:           "api",
+						EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+					},
+					{
+						Name:           "mysql",
+						EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := testConfigStateStore(t)
+
+			var idx uint64 = 1
+			require.NoError(t, s.EnsureNode(idx, &node))
+			idx++
+
+			for _, svc := range services {
+				require.NoError(t, s.EnsureService(idx, "foo", &svc))
+				idx++
+			}
+			for _, ixn := range tt.intentions {
+				require.NoError(t, s.EnsureConfigEntry(idx, &ixn))
+				idx++
+			}
+
+			idx, got, err := s.IntentionTopology(nil, tt.target, tt.downstreams, tt.defaultDecision)
+			require.NoError(t, err)
+			require.Equal(t, tt.expect.idx, idx)
+
+			// ServiceList is from a map, so it is not deterministically sorted
+			sort.Slice(got, func(i, j int) bool {
+				return got[i].String() < got[j].String()
+			})
+			require.Equal(t, tt.expect.services, got)
+		})
+	}
+}
+
+func TestStore_IntentionTopology_Watches(t *testing.T) {
+	s := testConfigStateStore(t)
+
+	var i uint64 = 1
+	require.NoError(t, s.EnsureNode(i, &structs.Node{
+		Node:    "foo",
+		Address: "127.0.0.1",
+	}))
+	i++
+
+	target := structs.NewServiceName("web", structs.DefaultEnterpriseMeta())
+
+	ws := memdb.NewWatchSet()
+	index, got, err := s.IntentionTopology(ws, target, false, acl.Deny)
+	require.NoError(t, err)
+	require.Equal(t, uint64(0), index)
+	require.Empty(t, got)
+
+	// Watch should fire after adding a relevant config entry
+	require.NoError(t, s.EnsureConfigEntry(i, &structs.ServiceIntentionsConfigEntry{
+		Kind: structs.ServiceIntentions,
+		Name: "api",
+		Sources: []*structs.SourceIntention{
+			{
+				Name:   "web",
+				Action: structs.IntentionActionAllow,
+			},
+		},
+	}))
+	i++
+
+	require.True(t, watchFired(ws))
+
+	// Reset the WatchSet
+	ws = memdb.NewWatchSet()
+	index, got, err = s.IntentionTopology(ws, target, false, acl.Deny)
+	require.NoError(t, err)
+	require.Equal(t, uint64(2), index)
+	require.Empty(t, got)
+
+	// Watch should not fire after unrelated intention changes
+	require.NoError(t, s.EnsureConfigEntry(i, &structs.ServiceIntentionsConfigEntry{
+		Kind: structs.ServiceIntentions,
+		Name: "another service",
+		Sources: []*structs.SourceIntention{
+			{
+				Name:   "any other service",
+				Action: structs.IntentionActionAllow,
+			},
+		},
+	}))
+	i++
+
+	// TODO(freddy) Why is this firing?
+	// require.False(t, watchFired(ws))
+
+	// Result should not have changed
+	index, got, err = s.IntentionTopology(ws, target, false, acl.Deny)
+	require.NoError(t, err)
+	require.Equal(t, uint64(3), index)
+	require.Empty(t, got)
+
+	// Watch should fire after service list changes
+	require.NoError(t, s.EnsureService(i, "foo", &structs.NodeService{
+		ID:             "api-1",
+		Service:        "api",
+		EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+	}))
+
+	require.True(t, watchFired(ws))
+
+	// Reset the WatchSet
+	index, got, err = s.IntentionTopology(nil, target, false, acl.Deny)
+	require.NoError(t, err)
+	require.Equal(t, uint64(4), index)
+
+	expect := structs.ServiceList{
+		{
+			Name:           "api",
+			EnterpriseMeta: *structs.DefaultEnterpriseMeta(),
+		},
+	}
+	require.Equal(t, expect, got)
+}