peering, state: account for peer intentions (#13443)

Signed-off-by: acpana <8968914+acpana@users.noreply.github.com>

agent/consul/intention_endpoint.go

@@ -432,7 +432,7 @@ func (s *Intention) Get(args *structs.IntentionQueryRequest, reply *structs.Inde
 	}
 
 	if args.Exact != nil {
-		// // Finish defaulting the namespace fields.
+		// Finish defaulting the namespace fields.
 		if args.Exact.SourceNS == "" {
 			args.Exact.SourceNS = entMeta.NamespaceOrDefault()
 		}

agent/consul/state/config_entry_intention.go

@@ -202,10 +202,13 @@ func (s *Store) configIntentionGetExactTxn(tx ReadTxn, ws memdb.WatchSet, args *
 		return idx, nil, nil, nil
 	}
 
-	sn := structs.NewServiceName(args.SourceName, args.SourceEnterpriseMeta())
+	psn := structs.PeeredServiceName{
+		Peer:        args.SourcePeer,
+		ServiceName: structs.NewServiceName(args.SourceName, args.SourceEnterpriseMeta()),
+	}
 
 	for _, src := range entry.Sources {
-		if sn == src.SourceServiceName() {
+		if psn.Peer == src.Peer && psn.ServiceName == src.SourceServiceName() {
 			return idx, entry, entry.ToIntention(src), nil
 		}
 	}

agent/consul/state/intention_test.go

@@ -1199,6 +1199,72 @@ func TestStore_IntentionsList(t *testing.T) {
 	})
 }
 
+// TestStore_IntentionExact_ConfigEntries test that we can get a local config entry intention
+// and a peered config entry intention with an IntentionGetExact call
+func TestStore_IntentionExact_ConfigEntries(t *testing.T) {
+	s := testConfigStateStore(t)
+	inputs := []*structs.ServiceIntentionsConfigEntry{
+		{
+			Kind: structs.ServiceIntentions,
+			Name: "foo",
+			Sources: []*structs.SourceIntention{
+				{
+					Action:      structs.IntentionActionAllow,
+					Name:        "bar",
+					Peer:        "peer1",
+					Description: "peered service intention",
+				},
+				{
+					Action:      structs.IntentionActionAllow,
+					Name:        "bar",
+					Description: "local service intention",
+				},
+			},
+		},
+	}
+	idx := uint64(0)
+
+	for _, input := range inputs {
+		require.NoError(t, input.Normalize())
+		require.NoError(t, input.Validate())
+		idx++
+		require.NoError(t, s.EnsureConfigEntry(idx, input))
+	}
+
+	t.Run("assert that we can get the peered intention", func(t *testing.T) {
+		idx, entry, ixn, err := s.IntentionGetExact(nil, &structs.IntentionQueryExact{
+			SourceNS:        "default",
+			SourceName:      "bar",
+			SourcePeer:      "peer1",
+			DestinationNS:   "default",
+			DestinationName: "foo",
+		})
+
+		require.NoError(t, err)
+		require.NotNil(t, entry)
+		require.NotNil(t, ixn)
+		require.Equal(t, "peer1", ixn.SourcePeer)
+		require.Equal(t, "peered service intention", ixn.Description)
+		require.Equal(t, uint64(1), idx)
+	})
+
+	t.Run("assert that we can get the local intention", func(t *testing.T) {
+		idx, entry, ixn, err := s.IntentionGetExact(nil, &structs.IntentionQueryExact{
+			SourceNS:        "default",
+			SourceName:      "bar",
+			DestinationNS:   "default",
+			DestinationName: "foo",
+		})
+
+		require.NoError(t, err)
+		require.NotNil(t, entry)
+		require.NotNil(t, ixn)
+		require.Empty(t, ixn.SourcePeer)
+		require.Equal(t, "local service intention", ixn.Description)
+		require.Equal(t, uint64(1), idx)
+	})
+}
+
 func TestStore_IntentionMatch_ConfigEntries(t *testing.T) {
 	type testcase struct {
 		name   string

agent/structs/config_entry_intentions.go

@@ -573,7 +573,7 @@ func (e *ServiceIntentionsConfigEntry) validate(legacyWrite bool) error {
 		return fmt.Errorf("At least one source is required")
 	}
 
-	seenSources := make(map[ServiceName]struct{})
+	seenSources := make(map[PeeredServiceName]struct{})
 	for i, src := range e.Sources {
 		if src.Name == "" {
 			return fmt.Errorf("Sources[%d].Name is required", i)
@@ -761,11 +761,15 @@ func (e *ServiceIntentionsConfigEntry) validate(legacyWrite bool) error {
 			}
 		}
 
-		serviceName := src.SourceServiceName()
+		psn := PeeredServiceName{Peer: src.Peer, ServiceName: src.SourceServiceName()}
-		if _, exists := seenSources[serviceName]; exists {
+		if _, exists := seenSources[psn]; exists {
-			return fmt.Errorf("Sources[%d] defines %q more than once", i, serviceName.String())
+			if psn.Peer != "" {
+				return fmt.Errorf("Sources[%d] defines peer(%q) %q more than once", i, psn.Peer, psn.ServiceName.String())
+			} else {
+				return fmt.Errorf("Sources[%d] defines %q more than once", i, psn.ServiceName.String())
 			}
-		seenSources[serviceName] = struct{}{}
+		}
+		seenSources[psn] = struct{}{}
 	}
 
 	return nil

agent/structs/config_entry_intentions_test.go

@@ -1226,6 +1226,42 @@ func TestServiceIntentionsConfigEntry(t *testing.T) {
 				}, entry.Sources)
 			},
 		},
+		"local and peer intentions are different": {
+			entry: &ServiceIntentionsConfigEntry{
+				Kind: ServiceIntentions,
+				Name: "test",
+				Sources: []*SourceIntention{
+					{
+						Name:   "foo",
+						Action: IntentionActionAllow,
+					},
+					{
+						Name:   "foo",
+						Peer:   "peer1",
+						Action: IntentionActionAllow,
+					},
+				},
+			},
+		},
+		"already have a peer intention for source": {
+			entry: &ServiceIntentionsConfigEntry{
+				Kind: ServiceIntentions,
+				Name: "test",
+				Sources: []*SourceIntention{
+					{
+						Name:   "foo",
+						Peer:   "peer1",
+						Action: IntentionActionAllow,
+					},
+					{
+						Name:   "foo",
+						Peer:   "peer1",
+						Action: IntentionActionAllow,
+					},
+				},
+			},
+			validateErr: `Sources[1] defines peer("peer1") "` + fooName.String() + `" more than once`,
+		},
 	}
 	for name, tc := range cases {
 		tc := tc