fix(policy): use new compute for rules and fix rules intersect (backp…

…ort of #12340) (#12516)

## Motivation

1. When I went through the policy matching subset `Intersect` method, I
found the behavior is incorrect with one case
2. Rewrite the rule's `Compute` method to use the new `ContainsElement`
method instead of using the old `IsSubset` method

## Implementation information

Introduce one new type `type Element map[string]string` for
differentiation by `IsSubset`.

<!--
> Changelog: skip
-->
<!--
Uncomment the above section to explicitly set a [`> Changelog:` entry
here](https://github.com/kumahq/kuma/blob/master/CONTRIBUTING.md#submitting-a-patch)?
-->

## Backport

cherry-picked commit 3a00363

---------

Signed-off-by: Icarus Wu <[email protected]>
Signed-off-by: Ilya Lobkov <[email protected]>
Co-authored-by: Ilya Lobkov <[email protected]>

pkg/core/xds/inspect/rules.go

@@ -112,8 +112,7 @@ func getOutboundRuleAttachments(rules core_rules.Rules, networking *mesh_proto.D
 		}
 		attachment := byUniqueClusterName[name]
 		if attachment == nil {
-			subset := core_rules.SubsetFromTags(outboundTags)
-			computedRule := rules.Compute(subset)
+			computedRule := rules.Compute(core_rules.Element(outboundTags))
 			if computedRule == nil {
 				continue
 			}

pkg/plugins/policies/core/rules/rules.go

@@ -118,6 +118,66 @@ type Tag struct {
 // Subset represents a group of proxies
 type Subset []Tag
 
+// ContainsElement returns true if there exists a key in 'other' that matches the current set,
+// and the corresponding k-v pair must match the set rule. Also, the left set rules of the current set can't make an impact.
+// Empty set is a superset for all elements.
+//
+// For example if you have a Subset with Tags: [{key: zone, value: east, not: true}, {key: service, value: frontend, not: false}]
+// an Element with k-v pairs: 1) service: frontend  2) version: zone1
+// there's a k-v pair 'service: frontend' in Element that matches the set rule {key: service, value: frontend, not: false}
+// the left set rule of Subset {key: zone, value: east, not: true} won't make an impact because of 'not: true'
+func (ss Subset) ContainsElement(other Element) bool {
+	// 1. find the overlaps of element and current subset
+	// 2. verify the overlaps
+	// 3. verify the left of current subset
+	// 4. if no overlaps, verify if all the Subset rules are negative
+
+	if len(ss) == 0 {
+		return true
+	}
+	if len(other) == 0 {
+		return false
+	}
+
+	hasOverlapKey := false
+	for _, tag := range ss {
+		otherVal, ok := other[tag.Key]
+		if ok {
+			hasOverlapKey = true
+
+			// contradict
+			if tag.Value == otherVal && tag.Not {
+				return false
+			}
+			// intersect
+			if tag.Value == otherVal && !tag.Not {
+				continue
+			}
+			// intersect
+			if tag.Value != otherVal && tag.Not {
+				continue
+			}
+			// contradict
+			if tag.Value != otherVal && !tag.Not {
+				return false
+			}
+		} else if !tag.Not {
+			// For those items that don't exist in element should not make an impact.
+			// For example, the DP with tag {"service: frontend"} doesn't match
+			// the policy with matching tags [{"service: frontend"}, {"zone": "east"}]
+			return false
+		}
+	}
+
+	// if the current Subset owns all of negative rules and no overlapped keys in Element,
+	// we can also regard the Subset contains Element
+	if !hasOverlapKey && ss.NumPositive() == 0 {
+		return true
+	}
+
+	return hasOverlapKey
+}
+
 // IsSubset returns true if 'other' is a subset of the current set.
 // Empty set is a superset for all subsets.
 func (ss Subset) IsSubset(other Subset) bool {
@@ -167,6 +227,36 @@ func isSubset(t1, t2 Tag) bool {
 
 // Intersect returns true if there exists an element that belongs both to 'other' and current set.
 // Empty set intersects with all sets.
+//
+// We're using this function to check if 2 'from' rules of MeshTrafficPermission can be applied to the same client DPP.
+// For example:
+//
+//	from:
+//	  - targetRef:
+//	      kind: MeshSubset
+//	      tags:
+//	        team: team-a
+//	  - targetRef:
+//	      kind: MeshSubset
+//	      tags:
+//	        zone: east
+//
+// there is a DPP with tags 'team: team-a' and 'zone: east' that's subjected to both these rules.
+// So 'from[0]' and 'from[1]' have an intersection.
+// However, in another example:
+//
+//	from:
+//	 - targetRef:
+//	     kind: MeshSubset
+//	     tags:
+//	       team: team-a
+//	 - targetRef:
+//	     kind: MeshSubset
+//	     tags:
+//	       team: team-b
+//	       zone: east
+//
+// there is no DPP that'd hit both 'from[0]' and 'from[1]'. So in this case they don't have an intersection.
 func (ss Subset) Intersect(other Subset) bool {
 	if len(ss) == 0 || len(other) == 0 {
 		return true
@@ -184,7 +274,7 @@ func (ss Subset) Intersect(other Subset) bool {
 		}
 		oTags, ok := otherByKeysOnlyPositive[tag.Key]
 		if !ok {
-			return true
+			continue
 		}
 		for _, otherTag := range oTags {
 			if otherTag != tag {
@@ -223,6 +313,26 @@ func SubsetFromTags(tags map[string]string) Subset {
 	return subset
 }
 
+type Element map[string]string
+
+func (e Element) WithKeyValue(key, value string) Element {
+	c := maps.Clone(e)
+	if c == nil {
+		c = Element{}
+	}
+
+	c[key] = value
+	return c
+}
+
+func MeshElement() Element {
+	return Element{}
+}
+
+func MeshServiceElement(name string) Element {
+	return Element{mesh_proto.ServiceTag: name}
+}
+
 // NumPositive returns a number of tags without negation
 func (ss Subset) NumPositive() int {
 	pos := 0
@@ -254,20 +364,23 @@ type Rule struct {
 
 type Rules []*Rule
 
-// Compute returns configuration for the given subset.
-func (rs Rules) Compute(sub Subset) *Rule {
+// Compute returns Rule for the given element.
+func (rs Rules) Compute(element Element) *Rule {
 	for _, rule := range rs {
-		if rule.Subset.IsSubset(sub) {
+		if rule.Subset.ContainsElement(element) {
 			return rule
 		}
 	}
 	return nil
 }
 
-func ComputeConf[T any](rs Rules, sub Subset) *T {
-	if computed := rs.Compute(sub); computed != nil {
+// ComputeConf returns configuration for the given element.
+func ComputeConf[T any](rs Rules, element Element) *T {
+	computed := rs.Compute(element)
+	if computed != nil {
 		return pointer.To(computed.Conf.(T))
 	}
+
 	return nil
 }