Handle Traffic Permissions With Empty Sources Properly (#19024)

Fix issues with empty sources

* Validate that each permission on traffic permissions resources has at least one source.
* Don't construct RBAC policies when there aren't any principals. This resulted in Envoy rejecting xDS updates with a validation error.

```
error=
  | rpc error: code = Internal desc = Error adding/updating listener(s) public_listener: Proto constraint validation failed (RBACValidationError.Rules: embedded message failed validation | caused by RBACValidationError.Policies[consul-intentions-layer4-1]: embedded message failed validation | caused by PolicyValidationError.Principals: value must contain at least 1 item(s)): rules {
```

agent/xds/rbac_test.go

@@ -768,6 +768,17 @@ func TestMakeRBACNetworkAndHTTPFilters(t *testing.T) {
 				},
 			},
 		},
+		// This validates that we don't send xDS messages to Envoy that will fail validation.
+		// Traffic permissions validations prevent this from being written to the IR, so the thing
+		// that matters is that the snapshot is valid to Envoy.
+		"v2-ignore-empty-permissions": {
+			intentionDefaultAllow: true,
+			v2L4TrafficPermissions: &pbproxystate.TrafficPermissions{
+				DenyPermissions: []*pbproxystate.Permission{
+					{},
+				},
+			},
+		},
 		"default-allow-kitchen-sink": {
 			intentionDefaultAllow: true,
 			v1Intentions: sorted(

agent/xds/testdata/rbac/v2-ignore-empty-permissions.golden

@@ -0,0 +1,22 @@
+{
+  "filters": [
+    {
+      "name": "envoy.filters.network.rbac",
+      "typedConfig": {
+        "@type": "type.googleapis.com/envoy.extensions.filters.network.rbac.v3.RBAC",
+        "rules": {
+          "action": "DENY"
+        },
+        "statPrefix": "connect_authz"
+      }
+    },
+    {
+      "name": "envoy.filters.network.rbac",
+      "typedConfig": {
+        "@type": "type.googleapis.com/envoy.extensions.filters.network.rbac.v3.RBAC",
+        "rules": {},
+        "statPrefix": "connect_authz"
+      }
+    }
+  ]
+}

agent/xdsv2/rbac_resources.go

@@ -84,13 +84,20 @@ func makeRBACPolicies(l4Permissions []*pbproxystate.Permission) map[string]*envo
 	policies := make(map[string]*envoy_rbac_v3.Policy, len(l4Permissions))
 
 	for i, permission := range l4Permissions {
-		policies[policyLabel(i)] = makeRBACPolicy(permission)
+		policy := makeRBACPolicy(permission)
+		if policy != nil {
+			policies[policyLabel(i)] = policy
+		}
 	}
 
 	return policies
 }
 
 func makeRBACPolicy(p *pbproxystate.Permission) *envoy_rbac_v3.Policy {
+	if len(p.Principals) == 0 {
+		return nil
+	}
+
 	var principals []*envoy_rbac_v3.Principal
 
 	for _, p := range p.Principals {

internal/auth/internal/controllers/trafficpermissions/controller_test.go

@@ -81,7 +81,15 @@ func (suite *controllerSuite) TestReconcile_CTPCreate_NoReferencingTrafficPermis
 
 func (suite *controllerSuite) TestReconcile_CTPCreate_ReferencingTrafficPermissionsExist() {
 	// create dead-end traffic permissions
-	p1 := &pbauth.Permission{}
+	p1 := &pbauth.Permission{
+		Sources: []*pbauth.Source{
+			{
+				IdentityName: "foo",
+				Namespace:    "default",
+				Partition:    "default",
+				Peer:         "local",
+			}},
+	}
 	tp1 := rtest.Resource(pbauth.TrafficPermissionsType, "tp1").WithData(suite.T(), &pbauth.TrafficPermissions{
 		Destination: &pbauth.Destination{
 			IdentityName: "wi1",
@@ -127,7 +135,15 @@ func (suite *controllerSuite) TestReconcile_CTPCreate_ReferencingTrafficPermissi
 }
 
 func (suite *controllerSuite) TestReconcile_WorkloadIdentityDelete_ReferencingTrafficPermissionsExist() {
-	p1 := &pbauth.Permission{}
+	p1 := &pbauth.Permission{
+		Sources: []*pbauth.Source{
+			{
+				IdentityName: "foo",
+				Namespace:    "default",
+				Partition:    "default",
+				Peer:         "local",
+			}},
+	}
 	tp1 := rtest.Resource(pbauth.TrafficPermissionsType, "tp1").WithData(suite.T(), &pbauth.TrafficPermissions{
 		Destination: &pbauth.Destination{
 			IdentityName: "wi1",
@@ -203,7 +219,15 @@ func (suite *controllerSuite) TestReconcile_TrafficPermissionsCreate_Destination
 	require.NoError(suite.T(), err)
 
 	// create traffic permissions
-	p1 := &pbauth.Permission{}
+	p1 := &pbauth.Permission{
+		Sources: []*pbauth.Source{
+			{
+				IdentityName: "foo",
+				Namespace:    "default",
+				Partition:    "default",
+				Peer:         "local",
+			}},
+	}
 	tp1 := rtest.Resource(pbauth.TrafficPermissionsType, "tp1").WithData(suite.T(), &pbauth.TrafficPermissions{
 		Destination: &pbauth.Destination{
 			IdentityName: "wi1",
@@ -275,7 +299,15 @@ func (suite *controllerSuite) TestReconcile_TrafficPermissionsDelete_Destination
 	require.NoError(suite.T(), err)
 
 	// create traffic permissions
-	p1 := &pbauth.Permission{}
+	p1 := &pbauth.Permission{
+		Sources: []*pbauth.Source{
+			{
+				IdentityName: "foo",
+				Namespace:    "default",
+				Partition:    "default",
+				Peer:         "local",
+			}},
+	}
 	tp1 := rtest.Resource(pbauth.TrafficPermissionsType, "tp1").WithData(suite.T(), &pbauth.TrafficPermissions{
 		Destination: &pbauth.Destination{
 			IdentityName: "wi1",
@@ -327,7 +359,15 @@ func (suite *controllerSuite) TestReconcile_TrafficPermissionsDelete_Destination
 
 func (suite *controllerSuite) TestReconcile_TrafficPermissionsDelete_DestinationWorkloadIdentityDoesNotExist() {
 	// create traffic permissions
-	p1 := &pbauth.Permission{}
+	p1 := &pbauth.Permission{
+		Sources: []*pbauth.Source{
+			{
+				IdentityName: "foo",
+				Namespace:    "default",
+				Partition:    "default",
+				Peer:         "local",
+			}},
+	}
 	tp1 := rtest.Resource(pbauth.TrafficPermissionsType, "tp1").WithData(suite.T(), &pbauth.TrafficPermissions{
 		Destination: &pbauth.Destination{
 			IdentityName: "wi1",

internal/auth/internal/types/traffic_permissions.go

@@ -152,6 +152,13 @@ func ValidateTrafficPermissions(res *pbresource.Resource) error {
 func validatePermission(p *pbauth.Permission, wrapErr func(error) error) error {
 	var merr error
 
+	if len(p.Sources) == 0 {
+		merr = multierror.Append(merr, wrapErr(resource.ErrInvalidField{
+			Name:    "sources",
+			Wrapped: resource.ErrEmpty,
+		}))
+	}
+
 	for s, src := range p.Sources {
 		wrapSrcErr := func(err error) error {
 			return wrapErr(resource.ErrInvalidListElement{

internal/auth/internal/types/traffic_permissions_test.go

@@ -124,7 +124,8 @@ type permissionTestCase struct {
 func permissionsTestCases() map[string]permissionTestCase {
 	return map[string]permissionTestCase{
 		"empty": {
-			p: &pbauth.Permission{},
+			p:         &pbauth.Permission{},
+			expectErr: `invalid "sources" field: cannot be empty`,
 		},
 		"empty-sources": {
 			p: &pbauth.Permission{
@@ -143,6 +144,7 @@ func permissionsTestCases() map[string]permissionTestCase {
 					},
 				},
 			},
+			expectErr: `invalid "sources" field: cannot be empty`,
 		},
 		"empty-destination-rules": {
 			p: &pbauth.Permission{