Net 6774 Make Sameness Groups Work With Traffic Permissions CE (#20316)

* Make Sameness Groups Work With Traffic Permissions

* Fix controller dependency

internal/auth/internal/controllers/register.go

@@ -5,6 +5,7 @@ package controllers
 
 import (
 	"github.com/hashicorp/consul/internal/auth/internal/controllers/trafficpermissions"
+	"github.com/hashicorp/consul/internal/auth/internal/controllers/trafficpermissions/expander"
 	"github.com/hashicorp/consul/internal/controller"
 )
 
@@ -13,5 +14,5 @@ type Dependencies struct {
 }
 
 func Register(mgr *controller.Manager, deps Dependencies) {
-	mgr.Register(trafficpermissions.Controller(deps.TrafficPermissionsMapper))
+	mgr.Register(trafficpermissions.Controller(deps.TrafficPermissionsMapper, expander.GetSamenessGroupExpander()))
 }

internal/auth/internal/controllers/trafficpermissions/builder.go

@@ -0,0 +1,87 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package trafficpermissions
+
+import (
+	"sort"
+
+	"github.com/hashicorp/consul/internal/auth/internal/controllers/trafficpermissions/expander"
+	"github.com/hashicorp/consul/internal/resource"
+	pbauth "github.com/hashicorp/consul/proto-public/pbauth/v2beta1"
+	pbmulticluster "github.com/hashicorp/consul/proto-public/pbmulticluster/v2beta1"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+)
+
+type trafficPermissionsBuilder struct {
+	missing            map[resource.ReferenceKey]missingSamenessGroupReferences
+	isDefault          bool
+	allowedPermissions []*pbauth.Permission
+	denyPermissions    []*pbauth.Permission
+	sgExpander         expander.SamenessGroupExpander
+	sgMap              map[string][]*pbmulticluster.SamenessGroupMember
+}
+
+type missingSamenessGroupReferences struct {
+	resource       *pbresource.Resource
+	samenessGroups []string
+}
+
+func newTrafficPermissionsBuilder(expander expander.SamenessGroupExpander, sgMap map[string][]*pbmulticluster.SamenessGroupMember) *trafficPermissionsBuilder {
+	return &trafficPermissionsBuilder{
+		sgMap:              sgMap,
+		missing:            make(map[resource.ReferenceKey]missingSamenessGroupReferences),
+		isDefault:          true,
+		sgExpander:         expander,
+		allowedPermissions: make([]*pbauth.Permission, 0),
+		denyPermissions:    make([]*pbauth.Permission, 0),
+	}
+}
+
+// track will use all associated Traffic Permissions to create new ComputedTrafficPermissions samenessGroupsForTrafficPermission
+func (tpb *trafficPermissionsBuilder) track(dec *resource.DecodedResource[*pbauth.TrafficPermissions]) {
+	missingSamenessGroups := tpb.sgExpander.Expand(dec.Data, tpb.sgMap)
+
+	if len(missingSamenessGroups) > 0 {
+		tpb.missing[resource.NewReferenceKey(dec.Id)] = missingSamenessGroupReferences{
+			resource:       dec.Resource,
+			samenessGroups: missingSamenessGroups,
+		}
+	}
+
+	tpb.isDefault = false
+
+	if dec.Data.Action == pbauth.Action_ACTION_ALLOW {
+		tpb.allowedPermissions = append(tpb.allowedPermissions, dec.Data.Permissions...)
+	} else {
+		tpb.denyPermissions = append(tpb.denyPermissions, dec.Data.Permissions...)
+	}
+}
+
+func (tpb *trafficPermissionsBuilder) build() (*pbauth.ComputedTrafficPermissions, map[resource.ReferenceKey]missingSamenessGroupReferences) {
+	return &pbauth.ComputedTrafficPermissions{
+		AllowPermissions: tpb.allowedPermissions,
+		DenyPermissions:  tpb.denyPermissions,
+		IsDefault:        tpb.isDefault,
+	}, tpb.missing
+}
+
+func missingForCTP(missing map[resource.ReferenceKey]missingSamenessGroupReferences) []string {
+	m := make(map[string]struct{})
+
+	for _, sgRefs := range missing {
+		for _, sg := range sgRefs.samenessGroups {
+			m[sg] = struct{}{}
+		}
+	}
+
+	out := make([]string, 0, len(m))
+
+	for sg := range m {
+		out = append(out, sg)
+	}
+
+	sort.Strings(out)
+
+	return out
+}

internal/auth/internal/controllers/trafficpermissions/controller.go

@@ -9,6 +9,7 @@ import (
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/anypb"
 
+	"github.com/hashicorp/consul/internal/auth/internal/controllers/trafficpermissions/expander"
 	"github.com/hashicorp/consul/internal/controller"
 	"github.com/hashicorp/consul/internal/controller/dependency"
 	"github.com/hashicorp/consul/internal/resource"
@@ -33,25 +34,41 @@ type TrafficPermissionsMapper interface {
 
 // Controller creates a controller for automatic ComputedTrafficPermissions management for
 // updates to WorkloadIdentity or TrafficPermission resources.
-func Controller(mapper TrafficPermissionsMapper) *controller.Controller {
+func Controller(mapper TrafficPermissionsMapper, sgExpander expander.SamenessGroupExpander) *controller.Controller {
 	if mapper == nil {
-		panic("No TrafficPermissionsMapper was provided to the TrafficPermissionsController constructor")
+		panic("TrafficPermissionsMapper is required for TrafficPermissionsController constructor")
+	}
+	if sgExpander == nil {
+		panic("SamenessGroupExpander is required for TrafficPermissionsController constructor")
 	}
 
-	return controller.NewController(StatusKey, pbauth.ComputedTrafficPermissionsType).
+	samenessGroupIndex := GetSamenessGroupIndex()
+
+	ctrl := controller.NewController(StatusKey, pbauth.ComputedTrafficPermissionsType).
 		WithWatch(pbauth.WorkloadIdentityType, dependency.ReplaceType(pbauth.ComputedTrafficPermissionsType)).
-		WithWatch(pbauth.TrafficPermissionsType, mapper.MapTrafficPermissions).
-		WithReconciler(&reconciler{mapper: mapper})
+		WithWatch(pbauth.TrafficPermissionsType, mapper.MapTrafficPermissions, samenessGroupIndex).
+		WithReconciler(&reconciler{mapper: mapper, sgExpander: sgExpander})
+
+	return registerEnterpriseControllerWatchers(ctrl)
 }
 
 type reconciler struct {
-	mapper TrafficPermissionsMapper
+	mapper     TrafficPermissionsMapper
+	sgExpander expander.SamenessGroupExpander
 }
 
 // Reconcile will reconcile one ComputedTrafficPermission (CTP) in response to some event.
-// Events include adding, modifying or deleting a WorkloadIdentity or TrafficPermission.
+// Events include adding, modifying or deleting a WorkloadIdentity or TrafficPermission or SamenessGroupType.
 func (r *reconciler) Reconcile(ctx context.Context, rt controller.Runtime, req controller.Request) error {
 	rt.Logger = rt.Logger.With("resource-id", req.ID, "controller", StatusKey)
+
+	ctpID := req.ID
+	oldCTPData, err := resource.GetDecodedResource[*pbauth.ComputedTrafficPermissions](ctx, rt.Client, ctpID)
+	if err != nil {
+		rt.Logger.Error("error retrieving computed permissions", "error", err)
+		return err
+	}
+
 	/*
 	 * A CTP ID could come in for a variety of reasons.
 	 * 1. workload identity create / delete: this results in the creation / deletion of a new CTP
@@ -62,7 +79,7 @@ func (r *reconciler) Reconcile(ctx context.Context, rt controller.Runtime, req c
 	 * CTPs are always generated from WorkloadIdentities, therefore the WI resource must already exist.
 	 * If it is missing, that means it was deleted.
 	 */
-	ctpID := req.ID
+
 	wi := resource.ReplaceType(pbauth.WorkloadIdentityType, ctpID)
 	workloadIdentity, err := resource.GetDecodedResource[*pbauth.WorkloadIdentity](ctx, rt.Client, wi)
 	if err != nil {
@@ -75,11 +92,6 @@ func (r *reconciler) Reconcile(ctx context.Context, rt controller.Runtime, req c
 	}
 
 	// Check if CTP exists:
-	oldCTPData, err := resource.GetDecodedResource[*pbauth.ComputedTrafficPermissions](ctx, rt.Client, ctpID)
-	if err != nil {
-		rt.Logger.Error("error retrieving computed permissions", "error", err)
-		return err
-	}
 	var oldResource *pbresource.Resource
 	var owner *pbresource.ID
 	if oldCTPData == nil {
@@ -91,47 +103,153 @@ func (r *reconciler) Reconcile(ctx context.Context, rt controller.Runtime, req c
 		owner = oldCTPData.Resource.Owner
 	}
 
-	// Part 2: Recompute a CTP from TP create / modify / delete, or create a new CTP from existing TPs:
-	latestTrafficPermissions, err := computeNewTrafficPermissions(ctx, rt, r.mapper, ctpID, oldResource)
+	sgMap, err := r.sgExpander.List(ctx, rt, req)
+
 	if err != nil {
-		rt.Logger.Error("error calculating computed permissions", "error", err)
+		rt.Logger.Error("error retrieving sameness groups", err.Error())
 		return err
 	}
 
-	if oldCTPData != nil && proto.Equal(oldCTPData.Data, latestTrafficPermissions) {
-		// there are no changes to the computed traffic permissions, and we can return early
+	trafficPermissionBuilder := newTrafficPermissionsBuilder(r.sgExpander, sgMap)
+	var tpResources []*pbresource.Resource
+
+	// Part 2: Recompute a CTP from TP create / modify / delete, or create a new CTP from existing TPs:
+	trackedTPs := r.mapper.GetTrafficPermissionsForCTP(ctpID)
+	if len(trackedTPs) > 0 {
+		rt.Logger.Trace("got tracked traffic permissions for CTP", "tps:", trackedTPs)
+	} else {
+		rt.Logger.Trace("found no tracked traffic permissions for CTP")
+	}
+
+	for _, t := range trackedTPs {
+		rsp, err := resource.GetDecodedResource[*pbauth.TrafficPermissions](ctx, rt.Client, resource.IDFromReference(t))
+		if err != nil {
+			rt.Logger.Error("error reading traffic permissions resource for computation", "error", err)
+			writeFailedStatus(ctx, rt, oldResource, resource.IDFromReference(t), err.Error())
+			return err
+		}
+		if rsp == nil {
+			rt.Logger.Trace("untracking deleted TrafficPermissions", "traffic-permissions-name", t.Name)
+			r.mapper.UntrackTrafficPermissions(resource.IDFromReference(t))
+			continue
+		}
+		trafficPermissionBuilder.track(rsp)
+		tpResources = append(tpResources, rsp.Resource)
+	}
+
+	latestComputedTrafficPermissions, missing := trafficPermissionBuilder.build()
+
+	if err != nil {
+		rt.Logger.Error("error expanding sameness groups", err.Error())
+		return err
+	}
+
+	newCTPResource := oldResource
+
+	allMissing := missingForCTP(missing)
+
+	if (oldCTPData == nil) || (!proto.Equal(oldCTPData.Data, latestComputedTrafficPermissions)) {
 		rt.Logger.Trace("no new computed traffic permissions")
+
+		// We can't short circuit here because we always need to update statuses.
+		newCTPData, err := anypb.New(latestComputedTrafficPermissions)
+		if err != nil {
+			rt.Logger.Error("error marshalling latest traffic permissions", "error", err)
+			writeFailedStatus(ctx, rt, oldResource, nil, err.Error())
+			return err
+		}
+		rt.Logger.Trace("writing computed traffic permissions")
+		rsp, err := rt.Client.Write(ctx, &pbresource.WriteRequest{
+			Resource: &pbresource.Resource{
+				Id:    req.ID,
+				Data:  newCTPData,
+				Owner: owner,
+			},
+		})
+		if err != nil {
+			rt.Logger.Error("error writing new computed traffic permissions", "error", err)
+			writeFailedStatus(ctx, rt, oldResource, nil, err.Error())
+			return err
+		} else {
+			rt.Logger.Trace("new computed traffic permissions were successfully written")
+		}
+		newCTPResource = rsp.Resource
+	}
+
+	if len(allMissing) > 0 {
+		return writeMissingSgStatuses(ctx, rt, req, allMissing, newCTPResource, missing, tpResources)
+	}
+
+	return writeComputedStatuses(ctx, rt, req, newCTPResource, latestComputedTrafficPermissions.IsDefault, tpResources)
+}
+
+func writeComputedStatuses(ctx context.Context, rt controller.Runtime, req controller.Request, ctpResource *pbresource.Resource, isDefault bool,
+	trackedTPs []*pbresource.Resource) error {
+	for _, tp := range trackedTPs {
+		err := writeStatusWithConditions(ctx, rt, tp,
+			[]*pbresource.Condition{ConditionComputedTrafficPermission()})
+		if err != nil {
+			return err
+		}
+	}
+	condition := ConditionComputed(req.ID.Name, isDefault)
+	return writeStatusWithConditions(ctx, rt, ctpResource, []*pbresource.Condition{condition})
+}
+
+func writeMissingSgStatuses(ctx context.Context, rt controller.Runtime, req controller.Request, allMissing []string, newCTPResource *pbresource.Resource,
+	missing map[resource.ReferenceKey]missingSamenessGroupReferences, tpResources []*pbresource.Resource) error {
+
+	condition := ConditionMissingSamenessGroup(req.ID.Tenancy.Partition, allMissing)
+
+	rt.Logger.Trace("writing missing sameness groups status")
+	err := writeStatusWithConditions(ctx, rt, newCTPResource, []*pbresource.Condition{condition})
+	if err != nil {
+		return err
+	}
+	//writing status to traffic permissions
+	for _, sgRefs := range missing {
+		if len(sgRefs.samenessGroups) == 0 {
+			err := writeStatusWithConditions(ctx, rt, sgRefs.resource,
+				[]*pbresource.Condition{ConditionComputedTrafficPermission()})
+			if err != nil {
+				return err
+			}
+			continue
+		}
+		conditionTp := ConditionMissingSamenessGroup(req.ID.Tenancy.Partition, sgRefs.samenessGroups)
+		err := writeStatusWithConditions(ctx, rt, sgRefs.resource, []*pbresource.Condition{conditionTp})
+		if err != nil {
+			return err
+		}
+	}
+	for _, trackedTp := range tpResources {
+		if _, ok := missing[resource.NewReferenceKey(trackedTp.Id)]; ok {
+			continue
+		}
+		err := writeStatusWithConditions(ctx, rt, trackedTp,
+			[]*pbresource.Condition{ConditionComputedTrafficPermission()})
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func writeStatusWithConditions(ctx context.Context, rt controller.Runtime, res *pbresource.Resource,
+	conditions []*pbresource.Condition) error {
+
+	newStatus := &pbresource.Status{
+		ObservedGeneration: res.Generation,
+		Conditions:         conditions,
+	}
+
+	if resource.EqualStatus(res.Status[StatusKey], newStatus, false) {
+		rt.Logger.Trace("old status is same as new status. skipping write", "resource", res.Id)
 		return nil
 	}
-	newCTPData, err := anypb.New(latestTrafficPermissions)
-	if err != nil {
-		rt.Logger.Error("error marshalling latest traffic permissions", "error", err)
-		writeFailedStatus(ctx, rt, oldResource, nil, err.Error())
-		return err
-	}
-	rt.Logger.Trace("writing computed traffic permissions")
-	rsp, err := rt.Client.Write(ctx, &pbresource.WriteRequest{
-		Resource: &pbresource.Resource{
-			Id:    req.ID,
-			Data:  newCTPData,
-			Owner: owner,
-		},
-	})
-	if err != nil || rsp.Resource == nil {
-		rt.Logger.Error("error writing new computed traffic permissions", "error", err)
-		writeFailedStatus(ctx, rt, oldResource, nil, err.Error())
-		return err
-	} else {
-		rt.Logger.Trace("new computed traffic permissions were successfully written")
-	}
-	newStatus := &pbresource.Status{
-		ObservedGeneration: rsp.Resource.Generation,
-		Conditions: []*pbresource.Condition{
-			ConditionComputed(req.ID.Name, latestTrafficPermissions.IsDefault),
-		},
-	}
-	_, err = rt.Client.WriteStatus(ctx, &pbresource.WriteStatusRequest{
-		Id:     rsp.Resource.Id,
+
+	_, err := rt.Client.WriteStatus(ctx, &pbresource.WriteStatusRequest{
+		Id:     res.Id,
 		Key:    StatusKey,
 		Status: newStatus,
 	})
@@ -142,55 +260,8 @@ func writeFailedStatus(ctx context.Context, rt controller.Runtime, ctp *pbresour
 	if ctp == nil {
 		return nil
 	}
-	newStatus := &pbresource.Status{
-		ObservedGeneration: ctp.Generation,
-		Conditions: []*pbresource.Condition{
-			ConditionFailedToCompute(ctp.Id.Name, tp.Name, errDetail),
-		},
+	conditions := []*pbresource.Condition{
+		ConditionFailedToCompute(ctp.Id.Name, tp.GetName(), errDetail),
 	}
-	_, err := rt.Client.WriteStatus(ctx, &pbresource.WriteStatusRequest{
-		Id:     ctp.Id,
-		Key:    StatusKey,
-		Status: newStatus,
-	})
-	return err
-}
-
-// computeNewTrafficPermissions will use all associated Traffic Permissions to create new ComputedTrafficPermissions data
-func computeNewTrafficPermissions(ctx context.Context, rt controller.Runtime, wm TrafficPermissionsMapper, ctpID *pbresource.ID, ctp *pbresource.Resource) (*pbauth.ComputedTrafficPermissions, error) {
-	// Part 1: Get all TPs that apply to workload identity
-	// Get already associated WorkloadIdentities/CTPs for reconcile requests:
-	trackedTPs := wm.GetTrafficPermissionsForCTP(ctpID)
-	if len(trackedTPs) > 0 {
-		rt.Logger.Trace("got tracked traffic permissions for CTP", "tps:", trackedTPs)
-	} else {
-		rt.Logger.Trace("found no tracked traffic permissions for CTP")
-	}
-	ap := make([]*pbauth.Permission, 0)
-	dp := make([]*pbauth.Permission, 0)
-	isDefault := true
-	for _, t := range trackedTPs {
-		rsp, err := resource.GetDecodedResource[*pbauth.TrafficPermissions](ctx, rt.Client, resource.IDFromReference(t))
-		if err != nil {
-			rt.Logger.Error("error reading traffic permissions resource for computation", "error", err)
-			writeFailedStatus(ctx, rt, ctp, resource.IDFromReference(t), err.Error())
-			return nil, err
-		}
-		if rsp == nil {
-			rt.Logger.Trace("untracking deleted TrafficPermissions", "traffic-permissions-name", t.Name)
-			wm.UntrackTrafficPermissions(resource.IDFromReference(t))
-			continue
-		}
-		isDefault = false
-		if rsp.Data.Action == pbauth.Action_ACTION_ALLOW {
-			ap = append(ap, rsp.Data.Permissions...)
-		} else {
-			dp = append(dp, rsp.Data.Permissions...)
-		}
-	}
-	return &pbauth.ComputedTrafficPermissions{
-		AllowPermissions: ap,
-		DenyPermissions:  dp,
-		IsDefault:        isDefault,
-	}, nil
+	return writeStatusWithConditions(ctx, rt, ctp, conditions)
 }

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

@@ -12,9 +12,11 @@ import (
 	"github.com/stretchr/testify/suite"
 
 	svctest "github.com/hashicorp/consul/agent/grpc-external/services/resource/testing"
+	"github.com/hashicorp/consul/internal/auth/internal/controllers/trafficpermissions/expander"
 	"github.com/hashicorp/consul/internal/auth/internal/mappers/trafficpermissionsmapper"
 	"github.com/hashicorp/consul/internal/auth/internal/types"
 	"github.com/hashicorp/consul/internal/controller"
+	"github.com/hashicorp/consul/internal/multicluster"
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/resourcetest"
 	rtest "github.com/hashicorp/consul/internal/resource/resourcetest"
@@ -32,27 +34,31 @@ type controllerSuite struct {
 	rt     controller.Runtime
 
 	mapper       *trafficpermissionsmapper.TrafficPermissionsMapper
+	sgExpander   expander.SamenessGroupExpander
 	reconciler   *reconciler
 	tenancies    []*pbresource.Tenancy
 	isEnterprise bool
 }
 
 func (suite *controllerSuite) SetupTest() {
+
 	suite.isEnterprise = versiontest.IsEnterprise()
 	suite.tenancies = resourcetest.TestTenancies()
 	suite.ctx = testutil.TestContext(suite.T())
 	client := svctest.NewResourceServiceBuilder().
-		WithRegisterFns(types.Register).
+		WithRegisterFns(types.Register, multicluster.RegisterTypes).
 		WithTenancies(suite.tenancies...).
 		Run(suite.T())
 
 	suite.client = rtest.NewClient(client)
+
 	suite.rt = controller.Runtime{
 		Client: suite.client,
 		Logger: testutil.Logger(suite.T()),
 	}
 	suite.mapper = trafficpermissionsmapper.New()
-	suite.reconciler = &reconciler{mapper: suite.mapper}
+	suite.sgExpander = expander.GetSamenessGroupExpander()
+	suite.reconciler = &reconciler{mapper: suite.mapper, sgExpander: suite.sgExpander}
 }
 
 func (suite *controllerSuite) requireTrafficPermissionsTracking(tp *pbresource.Resource, ids ...*pbresource.ID) {
@@ -173,6 +179,7 @@ func (suite *controllerSuite) TestReconcile_WorkloadIdentityDelete_ReferencingTr
 		}).
 			WithTenancy(tenancy).
 			Write(suite.T(), suite.client)
+
 		wi1ID := &pbresource.ID{
 			Name:    "wi1",
 			Type:    pbauth.ComputedTrafficPermissionsType,
@@ -323,7 +330,7 @@ func (suite *controllerSuite) TestReconcile_TrafficPermissionsCreate_Destination
 		rtest.RequireOwner(suite.T(), ctpResource, wi.Id, true)
 		assertCTPDefaultStatus(suite.T(), ctpResource, false)
 
-		// Delete the traffic permissions without updating the caches. Ensure is default is right even when the caches contain stale data.
+		// Delete the traffic permissions without updating the caches. Ensure is default is right even when the caches contain stale samenessGroupsForTrafficPermission.
 		suite.client.MustDelete(suite.T(), tp1.Id)
 		suite.client.MustDelete(suite.T(), tp2.Id)
 		suite.client.MustDelete(suite.T(), tp3.Id)
@@ -475,7 +482,7 @@ func (suite *controllerSuite) TestControllerBasic() {
 	// TODO: refactor this
 	// In this test we check basic operations for a workload identity and referencing traffic permission
 	mgr := controller.NewManager(suite.client, suite.rt.Logger)
-	mgr.Register(Controller(suite.mapper))
+	mgr.Register(Controller(suite.mapper, suite.sgExpander))
 	mgr.SetRaftLeader(true)
 	go mgr.Run(suite.ctx)
 
@@ -588,7 +595,7 @@ func (suite *controllerSuite) TestControllerBasicWithMultipleTenancyLevels() {
 	// TODO: refactor this
 	// In this test we check basic operations for a workload identity and referencing traffic permission
 	mgr := controller.NewManager(suite.client, suite.rt.Logger)
-	mgr.Register(Controller(suite.mapper))
+	mgr.Register(Controller(suite.mapper, suite.sgExpander))
 	mgr.SetRaftLeader(true)
 	go mgr.Run(suite.ctx)
 
@@ -709,7 +716,7 @@ func (suite *controllerSuite) TestControllerMultipleTrafficPermissions() {
 	suite.T().Skip("flaky behavior observed")
 	// In this test we check operations for a workload identity and multiple referencing traffic permissions
 	mgr := controller.NewManager(suite.client, suite.rt.Logger)
-	mgr.Register(Controller(suite.mapper))
+	mgr.Register(Controller(suite.mapper, suite.sgExpander))
 	mgr.SetRaftLeader(true)
 	go mgr.Run(suite.ctx)
 

internal/auth/internal/controllers/trafficpermissions/expander/expander_ce.go

@@ -0,0 +1,14 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+//go:build !consulent
+
+package expander
+
+import (
+	"github.com/hashicorp/consul/internal/auth/internal/controllers/trafficpermissions/expander/expander_ce"
+)
+
+func GetSamenessGroupExpander() *expander_ce.SamenessGroupExpander {
+	return expander_ce.New()
+}

internal/auth/internal/controllers/trafficpermissions/expander/expander_ce/expander_ce.go

@@ -0,0 +1,31 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package expander_ce
+
+import (
+	"context"
+
+	pbmulticluster "github.com/hashicorp/consul/proto-public/pbmulticluster/v2beta1"
+
+	"github.com/hashicorp/consul/internal/controller"
+	pbauth "github.com/hashicorp/consul/proto-public/pbauth/v2beta1"
+)
+
+type SamenessGroupExpander struct{}
+
+func New() *SamenessGroupExpander {
+	return &SamenessGroupExpander{}
+}
+
+func (sgE *SamenessGroupExpander) Expand(_ *pbauth.TrafficPermissions,
+	_ map[string][]*pbmulticluster.SamenessGroupMember) []string {
+	//no-op for CE
+	return nil
+}
+
+func (sgE *SamenessGroupExpander) List(_ context.Context, _ controller.Runtime,
+	_ controller.Request) (map[string][]*pbmulticluster.SamenessGroupMember, error) {
+	//no-op for CE
+	return nil, nil
+}

internal/auth/internal/controllers/trafficpermissions/expander/interface.go

@@ -0,0 +1,18 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package expander
+
+import (
+	"context"
+
+	"github.com/hashicorp/consul/internal/controller"
+	pbauth "github.com/hashicorp/consul/proto-public/pbauth/v2beta1"
+	pbmulticluster "github.com/hashicorp/consul/proto-public/pbmulticluster/v2beta1"
+)
+
+// SamenessgroupExpander is used to expand sameness group for a ComputedTrafficPermission resource
+type SamenessGroupExpander interface {
+	Expand(*pbauth.TrafficPermissions, map[string][]*pbmulticluster.SamenessGroupMember) []string
+	List(context.Context, controller.Runtime, controller.Request) (map[string][]*pbmulticluster.SamenessGroupMember, error)
+}

internal/auth/internal/controllers/trafficpermissions/helpers_ce.go

@@ -0,0 +1,32 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+//go:build !consulent
+
+package trafficpermissions
+
+import (
+	"github.com/hashicorp/consul/internal/controller"
+	"github.com/hashicorp/consul/internal/controller/cache/index"
+	"github.com/hashicorp/consul/internal/controller/cache/indexers"
+	"github.com/hashicorp/consul/internal/resource"
+	pbauth "github.com/hashicorp/consul/proto-public/pbauth/v2beta1"
+)
+
+const SgIndexName = "samenessGroupIndex"
+
+func registerEnterpriseControllerWatchers(ctrl *controller.Controller) *controller.Controller {
+	return ctrl
+}
+
+func GetSamenessGroupIndex() *index.Index {
+	return indexers.DecodedMultiIndexer(
+		SgIndexName,
+		index.ReferenceOrIDFromArgs,
+		func(r *resource.DecodedResource[*pbauth.TrafficPermissions]) (bool, [][]byte, error) {
+			//no - op for CE
+			return false, nil, nil
+
+		},
+	)
+}

internal/auth/internal/controllers/trafficpermissions/status.go

@@ -5,17 +5,21 @@ package trafficpermissions
 
 import (
 	"fmt"
+	"sort"
+	"strings"
 
 	"github.com/hashicorp/consul/proto-public/pbresource"
 )
 
 const (
-	StatusKey                           = "consul.io/traffic-permissions"
-	StatusTrafficPermissionsComputed    = "Traffic permissions have been computed"
-	StatusTrafficPermissionsNotComputed = "Traffic permissions have not been computed"
-	ConditionPermissionsAppliedMsg      = "Workload identity %s has new permissions"
-	ConditionNoPermissionsMsg           = "Workload identity %s has no permissions"
-	ConditionPermissionsFailedMsg       = "Unable to calculate new permission set for Workload identity %s"
+	StatusKey                                = "consul.io/traffic-permissions"
+	StatusTrafficPermissionsComputed         = "Traffic permissions have been computed"
+	StatusTrafficPermissionsNotComputed      = "Traffic permissions have not been computed"
+	StatusTrafficPermissionsProcessed        = "Traffic Permission processed successfully"
+	ConditionPermissionsAppliedMsg           = "Workload identity %s has new permissions"
+	ConditionNoPermissionsMsg                = "Workload identity %s has no permissions"
+	ConditionPermissionsFailedMsg            = "Unable to calculate new permission set for Workload identity %s"
+	ConditionMissingSamenessGroupInPartition = "Missing Sameness Groups names in partition(%s) - %s"
 )
 
 func ConditionComputed(workloadIdentity string, isDefault bool) *pbresource.Condition {
@@ -30,6 +34,14 @@ func ConditionComputed(workloadIdentity string, isDefault bool) *pbresource.Cond
 	}
 }
 
+func ConditionComputedTrafficPermission() *pbresource.Condition {
+	return &pbresource.Condition{
+		Type:    StatusTrafficPermissionsComputed,
+		State:   pbresource.Condition_STATE_TRUE,
+		Message: StatusTrafficPermissionsProcessed,
+	}
+}
+
 func ConditionFailedToCompute(workloadIdentity string, trafficPermissions string, errDetail string) *pbresource.Condition {
 	message := fmt.Sprintf(ConditionPermissionsFailedMsg, workloadIdentity)
 	if len(trafficPermissions) > 0 {
@@ -44,3 +56,13 @@ func ConditionFailedToCompute(workloadIdentity string, trafficPermissions string
 		Message: message,
 	}
 }
+
+func ConditionMissingSamenessGroup(partition string, missingSamenessGroups []string) *pbresource.Condition {
+	sort.Strings(missingSamenessGroups)
+	message := fmt.Sprintf(ConditionMissingSamenessGroupInPartition, partition, strings.Join(missingSamenessGroups, ","))
+	return &pbresource.Condition{
+		Type:    StatusTrafficPermissionsNotComputed,
+		State:   pbresource.Condition_STATE_FALSE,
+		Message: message,
+	}
+}

internal/auth/internal/types/computed_traffic_permissions.go

@@ -41,7 +41,7 @@ func validateComputedTrafficPermissions(res *DecodedComputedTrafficPermissions)
 				Wrapped: err,
 			}
 		}
-		if err := validatePermission(permission, wrapErr); err != nil {
+		if err := validatePermission(permission, res.Id, wrapErr); err != nil {
 			merr = multierror.Append(merr, err)
 		}
 	}
@@ -54,7 +54,7 @@ func validateComputedTrafficPermissions(res *DecodedComputedTrafficPermissions)
 				Wrapped: err,
 			}
 		}
-		if err := validatePermission(permission, wrapErr); err != nil {
+		if err := validatePermission(permission, res.Id, wrapErr); err != nil {
 			merr = multierror.Append(merr, err)
 		}
 	}

internal/auth/internal/types/traffic_permissions.go

@@ -137,7 +137,7 @@ func validateTrafficPermissions(res *DecodedTrafficPermissions) error {
 				Wrapped: err,
 			}
 		}
-		if err := validatePermission(permission, wrapErr); err != nil {
+		if err := validatePermission(permission, res.Id, wrapErr); err != nil {
 			merr = multierror.Append(merr, err)
 		}
 	}
@@ -145,7 +145,7 @@ func validateTrafficPermissions(res *DecodedTrafficPermissions) error {
 	return merr
 }
 
-func validatePermission(p *pbauth.Permission, wrapErr func(error) error) error {
+func validatePermission(p *pbauth.Permission, id *pbresource.ID, wrapErr func(error) error) error {
 	var merr error
 
 	if len(p.Sources) == 0 {
@@ -163,7 +163,7 @@ func validatePermission(p *pbauth.Permission, wrapErr func(error) error) error {
 				Wrapped: err,
 			})
 		}
-		if sourceHasIncompatibleTenancies(src) {
+		if sourceHasIncompatibleTenancies(src, id) {
 			merr = multierror.Append(merr, wrapSrcErr(resource.ErrInvalidListElement{
 				Name:    "source",
 				Wrapped: errSourcesTenancy,
@@ -194,7 +194,7 @@ func validatePermission(p *pbauth.Permission, wrapErr func(error) error) error {
 					Wrapped: err,
 				})
 			}
-			if sourceHasIncompatibleTenancies(d) {
+			if sourceHasIncompatibleTenancies(d, id) {
 				merr = multierror.Append(merr, wrapExclSrcErr(resource.ErrInvalidField{
 					Name:    "exclude_source",
 					Wrapped: errSourcesTenancy,
@@ -263,9 +263,12 @@ func validatePermission(p *pbauth.Permission, wrapErr func(error) error) error {
 	return merr
 }
 
-func sourceHasIncompatibleTenancies(src pbauth.SourceToSpiffe) bool {
+func sourceHasIncompatibleTenancies(src pbauth.SourceToSpiffe, id *pbresource.ID) bool {
+	if id.Tenancy == nil {
+		id.Tenancy = &pbresource.Tenancy{}
+	}
 	peerSet := src.GetPeer() != resource.DefaultPeerName
-	apSet := src.GetPartition() != resource.DefaultPartitionName
+	apSet := src.GetPartition() != id.Tenancy.Partition
 	sgSet := src.GetSamenessGroup() != ""
 
 	return (apSet && peerSet) || (apSet && sgSet) || (peerSet && sgSet)

internal/auth/internal/types/traffic_permissions_test.go

@@ -33,6 +33,7 @@ func TestValidateTrafficPermissions_ParseError(t *testing.T) {
 func TestValidateTrafficPermissions(t *testing.T) {
 	cases := map[string]struct {
 		tp        *pbauth.TrafficPermissions
+		id        *pbresource.ID
 		expectErr string
 	}{
 		"ok-minimal": {
@@ -161,13 +162,187 @@ func TestValidateTrafficPermissions(t *testing.T) {
 			},
 			expectErr: `invalid element at index 0 of list "permissions": invalid element at index 0 of list "destination_rules": invalid element at index 0 of list "destination_rule": traffic permissions with L7 rules are not yet supported`,
 		},
+		"source-has-same-tenancy-as-tp": {
+			id: &pbresource.ID{
+				Tenancy: &pbresource.Tenancy{
+					Partition: resource.DefaultPartitionName,
+				},
+			},
+			tp: &pbauth.TrafficPermissions{
+				Destination: &pbauth.Destination{
+					IdentityName: "w1",
+				},
+				Action: pbauth.Action_ACTION_ALLOW,
+				Permissions: []*pbauth.Permission{
+					{
+						Sources: []*pbauth.Source{
+							{
+								Partition:     resource.DefaultPartitionName,
+								Peer:          resource.DefaultPeerName,
+								SamenessGroup: "",
+							},
+						},
+					},
+				},
+			},
+		},
+		"source-has-partition-set": {
+			id: &pbresource.ID{
+				Tenancy: &pbresource.Tenancy{
+					Partition: resource.DefaultPartitionName,
+				},
+			},
+			tp: &pbauth.TrafficPermissions{
+				Destination: &pbauth.Destination{
+					IdentityName: "w1",
+				},
+				Action: pbauth.Action_ACTION_ALLOW,
+				Permissions: []*pbauth.Permission{
+					{
+						Sources: []*pbauth.Source{
+							{
+								Partition:     "part",
+								Peer:          resource.DefaultPeerName,
+								SamenessGroup: "",
+							},
+						},
+					},
+				},
+			},
+		},
+		"source-has-peer-set": {
+			id: &pbresource.ID{
+				Tenancy: &pbresource.Tenancy{
+					Partition: resource.DefaultPartitionName,
+				},
+			},
+			tp: &pbauth.TrafficPermissions{
+				Destination: &pbauth.Destination{
+					IdentityName: "w1",
+				},
+				Action: pbauth.Action_ACTION_ALLOW,
+				Permissions: []*pbauth.Permission{
+					{
+						Sources: []*pbauth.Source{
+							{
+								Partition:     resource.DefaultNamespaceName,
+								Peer:          "peer",
+								SamenessGroup: "",
+							},
+						},
+					},
+				},
+			},
+		},
+		"source-has-sameness-group-set": {
+			id: &pbresource.ID{
+				Tenancy: &pbresource.Tenancy{
+					Partition: resource.DefaultPartitionName,
+				},
+			},
+			tp: &pbauth.TrafficPermissions{
+				Destination: &pbauth.Destination{
+					IdentityName: "w1",
+				},
+				Action: pbauth.Action_ACTION_ALLOW,
+				Permissions: []*pbauth.Permission{
+					{
+						Sources: []*pbauth.Source{
+							{
+								Partition:     resource.DefaultNamespaceName,
+								Peer:          resource.DefaultPeerName,
+								SamenessGroup: "sg1",
+							},
+						},
+					},
+				},
+			},
+		},
+		"source-has-peer-and-partition-set": {
+			id: &pbresource.ID{
+				Tenancy: &pbresource.Tenancy{
+					Partition: resource.DefaultPartitionName,
+				},
+			},
+			tp: &pbauth.TrafficPermissions{
+				Destination: &pbauth.Destination{
+					IdentityName: "w1",
+				},
+				Action: pbauth.Action_ACTION_ALLOW,
+				Permissions: []*pbauth.Permission{
+					{
+						Sources: []*pbauth.Source{
+							{
+								Partition:     "part",
+								Peer:          "peer",
+								SamenessGroup: "",
+							},
+						},
+					},
+				},
+			},
+			expectErr: `invalid element at index 0 of list "permissions": invalid element at index 0 of list "sources": invalid element at index 0 of list "source": permissions sources may not specify partitions, peers, and sameness_groups together`,
+		},
+		"source-has-sameness-group-and-partition-set": {
+			id: &pbresource.ID{
+				Tenancy: &pbresource.Tenancy{
+					Partition: resource.DefaultPartitionName,
+				},
+			},
+			tp: &pbauth.TrafficPermissions{
+				Destination: &pbauth.Destination{
+					IdentityName: "w1",
+				},
+				Action: pbauth.Action_ACTION_ALLOW,
+				Permissions: []*pbauth.Permission{
+					{
+						Sources: []*pbauth.Source{
+							{
+								Partition:     "part",
+								Peer:          resource.DefaultPeerName,
+								SamenessGroup: "sg1",
+							},
+						},
+					},
+				},
+			},
+			expectErr: `invalid element at index 0 of list "permissions": invalid element at index 0 of list "sources": invalid element at index 0 of list "source": permissions sources may not specify partitions, peers, and sameness_groups together`,
+		},
+		"source-has-sameness-group-and-partition-peer-set": {
+			id: &pbresource.ID{
+				Tenancy: &pbresource.Tenancy{
+					Partition: resource.DefaultPartitionName,
+				},
+			},
+			tp: &pbauth.TrafficPermissions{
+				Destination: &pbauth.Destination{
+					IdentityName: "w1",
+				},
+				Action: pbauth.Action_ACTION_ALLOW,
+				Permissions: []*pbauth.Permission{
+					{
+						Sources: []*pbauth.Source{
+							{
+								Partition:     "part",
+								Peer:          "peer",
+								SamenessGroup: "sg1",
+							},
+						},
+					},
+				},
+			},
+			expectErr: `invalid element at index 0 of list "permissions": invalid element at index 0 of list "sources": invalid element at index 0 of list "source": permissions sources may not specify partitions, peers, and sameness_groups together`,
+		},
 	}
 
 	for n, tc := range cases {
 		t.Run(n, func(t *testing.T) {
-			res := resourcetest.Resource(pbauth.TrafficPermissionsType, "tp").
-				WithData(t, tc.tp).
-				Build()
+			resBuilder := resourcetest.Resource(pbauth.TrafficPermissionsType, "tp").
+				WithData(t, tc.tp)
+			if tc.id != nil {
+				resBuilder = resBuilder.WithTenancy(tc.id.Tenancy)
+			}
+			res := resBuilder.Build()
 
 			err := ValidateTrafficPermissions(res)
 			if tc.expectErr == "" {

proto-public/pbauth/v2beta1/traffic_permissions_helper.go

@@ -0,0 +1,15 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package authv2beta1
+
+func (ctp *TrafficPermissions) HasReferencedSamenessGroups() bool {
+	for _, dp := range ctp.Permissions {
+		for _, source := range dp.Sources {
+			if source.SamenessGroup != "" {
+				return true
+			}
+		}
+	}
+	return false
+}

proto-public/pbauth/v2beta1/traffic_permissions_helper_test.go

@@ -0,0 +1,50 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package authv2beta1
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestHasReferencedSamenessGroups(t *testing.T) {
+	type testCase struct {
+		tp       *TrafficPermissions
+		expected bool
+	}
+	testCases := []*testCase{
+		{
+			tp: &TrafficPermissions{
+				Permissions: []*Permission{
+					{
+						Sources: []*Source{
+							{
+								SamenessGroup: "sg1",
+							},
+						},
+					},
+				},
+			},
+			expected: true,
+		},
+		{
+			tp: &TrafficPermissions{
+				Permissions: []*Permission{
+					{
+						Sources: []*Source{
+							{
+								Peer: "peer",
+							},
+						},
+					},
+				},
+			},
+			expected: false,
+		},
+	}
+	for _, tc := range testCases {
+		require.Equal(t, tc.tp.HasReferencedSamenessGroups(), tc.expected)
+	}
+}