Update RBAC to handle imported services (#13404)

When converting from Consul intentions to xds RBAC rules, services imported from other peers must encode additional data like partition (from the remote cluster) and trust domain.

This PR updates the PeeringTrustBundle to hold the sending side's local partition as ExportedPartition. It also updates RBAC code to encode SpiffeIDs of imported services with the ExportedPartition and TrustDomain.

agent/connect/uri_service.go

@@ -1,6 +1,7 @@
 package connect
 
 import (
+	"fmt"
 	"net/url"
 
 	"github.com/hashicorp/consul/acl"
@@ -23,10 +24,6 @@ func (id SpiffeIDService) MatchesPartition(partition string) bool {
 	return id.PartitionOrDefault() == acl.PartitionOrDefault(partition)
 }
 
-func (id SpiffeIDService) PartitionOrDefault() string {
-	return acl.PartitionOrDefault(id.Partition)
-}
-
 // URI returns the *url.URL for this SPIFFE ID.
 func (id SpiffeIDService) URI() *url.URL {
 	var result url.URL
@@ -35,3 +32,20 @@ func (id SpiffeIDService) URI() *url.URL {
 	result.Path = id.uriPath()
 	return &result
 }
+
+func (id SpiffeIDService) uriPath() string {
+	path := fmt.Sprintf("/ns/%s/dc/%s/svc/%s",
+		id.NamespaceOrDefault(),
+		id.Datacenter,
+		id.Service,
+	)
+
+	// Although OSS has no support for partitions, it still needs to be able to
+	// handle exportedPartition from peered Consul Enterprise clusters in order
+	// to generate the correct SpiffeID.
+	// We intentionally avoid using pbpartition.DefaultName here to be OSS friendly.
+	if ap := id.PartitionOrDefault(); ap != "" && ap != "default" {
+		return "/ap/" + ap + path
+	}
+	return path
+}

agent/connect/uri_service_oss.go

@@ -4,7 +4,7 @@
 package connect
 
 import (
-	"fmt"
+	"strings"
 
 	"github.com/hashicorp/consul/acl"
 )
@@ -15,10 +15,14 @@ func (id SpiffeIDService) GetEnterpriseMeta() *acl.EnterpriseMeta {
 	return &acl.EnterpriseMeta{}
 }
 
-func (id SpiffeIDService) uriPath() string {
-	return fmt.Sprintf("/ns/%s/dc/%s/svc/%s",
-		id.NamespaceOrDefault(),
-		id.Datacenter,
-		id.Service,
-	)
+// PartitionOrDefault breaks from OSS's pattern of returning empty strings.
+// Although OSS has no support for partitions, it still needs to be able to
+// handle exportedPartition from peered Consul Enterprise clusters in order
+// to generate the correct SpiffeID.
+func (id SpiffeIDService) PartitionOrDefault() string {
+	if id.Partition == "" {
+		return "default"
+	}
+
+	return strings.ToLower(id.Partition)
 }

agent/connect/uri_service_oss_test.go

@@ -19,16 +19,6 @@ func TestSpiffeIDServiceURI(t *testing.T) {
 		require.Equal(t, "spiffe://1234.consul/ns/default/dc/dc1/svc/web", svc.URI().String())
 	})
 
-	t.Run("partitions are ignored", func(t *testing.T) {
-		svc := &SpiffeIDService{
-			Host:       "1234.consul",
-			Partition:  "other",
-			Datacenter: "dc1",
-			Service:    "web",
-		}
-		require.Equal(t, "spiffe://1234.consul/ns/default/dc/dc1/svc/web", svc.URI().String())
-	})
-
 	t.Run("namespaces are ignored", func(t *testing.T) {
 		svc := &SpiffeIDService{
 			Host:       "1234.consul",

agent/rpc/peering/subscription_manager.go

@@ -78,7 +78,7 @@ func (m *subscriptionManager) subscribe(ctx context.Context, peerID, peerName, p
 	if m.config.ConnectEnabled {
 		go m.notifyMeshGatewaysForPartition(ctx, state, state.partition)
 		// If connect is enabled, watch for updates to CA roots.
-		go m.notifyRootCAUpdates(ctx, state.updateCh)
+		go m.notifyRootCAUpdatesForPartition(ctx, state.updateCh, state.partition)
 	}
 
 	// This goroutine is the only one allowed to manipulate protected
@@ -291,14 +291,18 @@ func filterConnectReferences(orig *pbservice.IndexedCheckServiceNodes) {
 	orig.Nodes = newNodes
 }
 
-func (m *subscriptionManager) notifyRootCAUpdates(ctx context.Context, updateCh chan<- cache.UpdateEvent) {
+func (m *subscriptionManager) notifyRootCAUpdatesForPartition(
+	ctx context.Context,
+	updateCh chan<- cache.UpdateEvent,
+	partition string,
+) {
 	var idx uint64
 	// TODO(peering): retry logic; fail past a threshold
 	for {
 		var err error
 		// Typically, this function will block inside `m.subscribeCARoots` and only return on error.
 		// Errors are logged and the watch is retried.
-		idx, err = m.subscribeCARoots(ctx, idx, updateCh)
+		idx, err = m.subscribeCARoots(ctx, idx, updateCh, partition)
 		if errors.Is(err, stream.ErrSubForceClosed) {
 			m.logger.Trace("subscription force-closed due to an ACL change or snapshot restore, will attempt resume")
 		} else if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
@@ -317,7 +321,12 @@ func (m *subscriptionManager) notifyRootCAUpdates(ctx context.Context, updateCh
 
 // subscribeCARoots subscribes to state.EventTopicCARoots for changes to CA roots.
 // Upon receiving an event it will send the payload in updateCh.
-func (m *subscriptionManager) subscribeCARoots(ctx context.Context, idx uint64, updateCh chan<- cache.UpdateEvent) (uint64, error) {
+func (m *subscriptionManager) subscribeCARoots(
+	ctx context.Context,
+	idx uint64,
+	updateCh chan<- cache.UpdateEvent,
+	partition string,
+) (uint64, error) {
 	// following code adapted from connectca/watch_roots.go
 	sub, err := m.backend.Subscribe(&stream.SubscribeRequest{
 		Topic:   state.EventTopicCARoots,
@@ -382,8 +391,10 @@ func (m *subscriptionManager) subscribeCARoots(ctx context.Context, idx uint64,
 		updateCh <- cache.UpdateEvent{
 			CorrelationID: subCARoot,
 			Result: &pbpeering.PeeringTrustBundle{
-				TrustDomain: m.trustDomain,
-				RootPEMs:    rootPems,
+				TrustDomain:       m.trustDomain,
+				RootPEMs:          rootPems,
+				ExportedPartition: partition,
+				// TODO(peering): revisit decision not to validate datacenter in RBAC
 			},
 		}
 	}

agent/xds/listeners.go

@@ -695,6 +695,7 @@ func (s *ResourceGenerator) injectConnectFilters(cfgSnap *proxycfg.ConfigSnapsho
 	authzFilter, err := makeRBACNetworkFilter(
 		cfgSnap.ConnectProxy.Intentions,
 		cfgSnap.IntentionDefaultAllow,
+		cfgSnap.ConnectProxy.PeerTrustBundles,
 	)
 	if err != nil {
 		return err
@@ -963,6 +964,7 @@ func (s *ResourceGenerator) makeInboundListener(cfgSnap *proxycfg.ConfigSnapshot
 			httpAuthzFilter, err := makeRBACHTTPFilter(
 				cfgSnap.ConnectProxy.Intentions,
 				cfgSnap.IntentionDefaultAllow,
+				cfgSnap.ConnectProxy.PeerTrustBundles,
 			)
 			if err != nil {
 				return nil, err
@@ -1019,6 +1021,7 @@ func (s *ResourceGenerator) makeInboundListener(cfgSnap *proxycfg.ConfigSnapshot
 		filterOpts.httpAuthzFilter, err = makeRBACHTTPFilter(
 			cfgSnap.ConnectProxy.Intentions,
 			cfgSnap.IntentionDefaultAllow,
+			cfgSnap.ConnectProxy.PeerTrustBundles,
 		)
 		if err != nil {
 			return nil, err
@@ -1295,6 +1298,7 @@ func (s *ResourceGenerator) makeFilterChainTerminatingGateway(
 		authFilter, err := makeRBACNetworkFilter(
 			intentions,
 			cfgSnap.IntentionDefaultAllow,
+			nil, // TODO(peering): verify intentions w peers don't apply to terminatingGateway
 		)
 		if err != nil {
 			return nil, err
@@ -1319,6 +1323,7 @@ func (s *ResourceGenerator) makeFilterChainTerminatingGateway(
 		opts.httpAuthzFilter, err = makeRBACHTTPFilter(
 			intentions,
 			cfgSnap.IntentionDefaultAllow,
+			nil, // TODO(peering): verify intentions w peers don't apply to terminatingGateway
 		)
 		if err != nil {
 			return nil, err