fix(iam): persist policy tags and count attachments across all principal kinds (#1040)

* fix(iam): persist policy tags and count attachments across all principal kinds

CreatePolicy built its record with an empty tag list and never read input.Tags, and GetPolicy, ListPolicies and CreatePolicy's own response projected no Tags field at all. Both halves are fixed, matching the role path.

buildAttachmentCounts scanned only the users bucket, so a policy attached to a role or group counted as unattached. That is not just a misleading number: DeletePolicy gates its conflict check on the same count, so an attached policy could be deleted, leaving the principal holding a dangling ARN.

The count and the reverse lookup are the same traversal with the names kept or discarded, so both now derive from one buildPolicyAttachments over users, roles and groups. Nothing is stored, so no counter can drift from the attachments it counts.

ListEntitiesForPolicy is implemented on that traversal, honouring EntityFilter and PathPrefix.

Closes: mulga-fvhlc
Closes: mulga-9i6uh
Closes: mulga-ryu30
Closes: mulga-8ranb

* test(iam): account for ListEntitiesForPolicy in the cross-tenant and policy-ARN gates

The action resolves PolicyArn through the same branch as GetPolicy, so it
belongs with that helper's siblings; it was in neither gate, which is what
the accounting test exists to catch.

spinifex/gateway/iam.go

@@ -98,6 +98,9 @@ var iamActions = map[string]iamAction{
 	"DeletePolicy": iamHandler(func(accountID string, input *iam.DeletePolicyInput, gw *GatewayConfig) (any, error) {
 		return gateway_iam.DeletePolicy(accountID, input, gw.IAMService)
 	}),
+	"ListEntitiesForPolicy": iamHandler(func(accountID string, input *iam.ListEntitiesForPolicyInput, gw *GatewayConfig) (any, error) {
+		return gateway_iam.ListEntitiesForPolicy(accountID, input, gw.IAMService)
+	}),
 
 	// Policy attachment
 	"AttachUserPolicy": iamHandler(func(accountID string, input *iam.AttachUserPolicyInput, gw *GatewayConfig) (any, error) {

spinifex/gateway/iam/authz.go

@@ -97,14 +97,15 @@ var iamScopes = map[string]resourceScope{
 	"ListGroupPolicies":         existingScope(arn.IAMGroup, "GroupName"),
 
 	// Managed policies are normalized into the caller's trusted account.
-	"CreatePolicy":       createScope(arn.IAMPolicy, "PolicyName"),
-	"GetPolicy":          {source: sourcePolicyARN, nameField: "PolicyArn"},
-	"GetPolicyVersion":   {source: sourcePolicyARN, nameField: "PolicyArn"},
-	"ListPolicyVersions": {source: sourcePolicyARN, nameField: "PolicyArn"},
-	"DeletePolicy":       {source: sourcePolicyARN, nameField: "PolicyArn"},
-	"TagPolicy":          {source: sourcePolicyARN, nameField: "PolicyArn"},
-	"UntagPolicy":        {source: sourcePolicyARN, nameField: "PolicyArn"},
-	"ListPolicyTags":     {source: sourcePolicyARN, nameField: "PolicyArn"},
+	"CreatePolicy":          createScope(arn.IAMPolicy, "PolicyName"),
+	"GetPolicy":             {source: sourcePolicyARN, nameField: "PolicyArn"},
+	"GetPolicyVersion":      {source: sourcePolicyARN, nameField: "PolicyArn"},
+	"ListPolicyVersions":    {source: sourcePolicyARN, nameField: "PolicyArn"},
+	"DeletePolicy":          {source: sourcePolicyARN, nameField: "PolicyArn"},
+	"ListEntitiesForPolicy": {source: sourcePolicyARN, nameField: "PolicyArn"},
+	"TagPolicy":             {source: sourcePolicyARN, nameField: "PolicyArn"},
+	"UntagPolicy":           {source: sourcePolicyARN, nameField: "PolicyArn"},
+	"ListPolicyTags":        {source: sourcePolicyARN, nameField: "PolicyArn"},
 
 	// Instance profiles. Role operands are IAM condition-key values, not extra resources.
 	"CreateInstanceProfile":         createScope(arn.IAMInstanceProfile, "InstanceProfileName"),

spinifex/gateway/iam/authz_test.go

@@ -171,6 +171,7 @@ func TestResourceARNs_PolicyARNIsReanchoredOntoCallerAccount(t *testing.T) {
 		{"TagPolicy", &iam.TagPolicyInput{PolicyArn: aws.String(foreign)}},
 		{"UntagPolicy", &iam.UntagPolicyInput{PolicyArn: aws.String(foreign)}},
 		{"ListPolicyTags", &iam.ListPolicyTagsInput{PolicyArn: aws.String(foreign)}},
+		{"ListEntitiesForPolicy", &iam.ListEntitiesForPolicyInput{PolicyArn: aws.String(foreign)}},
 	}
 	for _, tt := range tests {
 		t.Run(tt.action, func(t *testing.T) {

spinifex/gateway/iam/handler_test.go

@@ -79,6 +79,10 @@ func (s *stubIAMService) DeletePolicy(_ string, _ *iam.DeletePolicyInput) (*iam.
 	return &iam.DeletePolicyOutput{}, nil
 }
 
+func (s *stubIAMService) ListEntitiesForPolicy(_ string, _ *iam.ListEntitiesForPolicyInput) (*iam.ListEntitiesForPolicyOutput, error) {
+	return &iam.ListEntitiesForPolicyOutput{}, nil
+}
+
 func (s *stubIAMService) AttachUserPolicy(_ string, _ *iam.AttachUserPolicyInput) (*iam.AttachUserPolicyOutput, error) {
 	return &iam.AttachUserPolicyOutput{}, nil
 }
@@ -568,6 +572,30 @@ func TestDeletePolicy(t *testing.T) {
 	}
 }
 
+func TestListEntitiesForPolicy(t *testing.T) {
+	svc := &stubIAMService{}
+	tests := []struct {
+		name    string
+		input   *iam.ListEntitiesForPolicyInput
+		wantErr string
+	}{
+		{"nil PolicyArn", &iam.ListEntitiesForPolicyInput{}, awserrors.ErrorMissingParameter},
+		{"empty PolicyArn", &iam.ListEntitiesForPolicyInput{PolicyArn: aws.String("")}, awserrors.ErrorMissingParameter},
+		{"valid", &iam.ListEntitiesForPolicyInput{PolicyArn: aws.String("arn:aws:iam::000000000000:policy/mypolicy")}, ""},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := ListEntitiesForPolicy(testAccountID, tc.input, svc)
+			if tc.wantErr != "" {
+				require.Error(t, err)
+				assert.Equal(t, tc.wantErr, err.Error())
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}
+
 // --- Policy Attachment ---
 
 func TestAttachUserPolicy(t *testing.T) {

spinifex/gateway/iam/policy.go

@@ -53,6 +53,13 @@ func DeletePolicy(accountID string, input *iam.DeletePolicyInput, svc handlers_i
 	return svc.DeletePolicy(accountID, input)
 }
 
+func ListEntitiesForPolicy(accountID string, input *iam.ListEntitiesForPolicyInput, svc handlers_iam.IAMService) (*iam.ListEntitiesForPolicyOutput, error) {
+	if input.PolicyArn == nil || *input.PolicyArn == "" {
+		return nil, errors.New(awserrors.ErrorMissingParameter)
+	}
+	return svc.ListEntitiesForPolicy(accountID, input)
+}
+
 func AttachUserPolicy(accountID string, input *iam.AttachUserPolicyInput, svc handlers_iam.IAMService) (*iam.AttachUserPolicyOutput, error) {
 	if input.UserName == nil || *input.UserName == "" {
 		return nil, errors.New(awserrors.ErrorMissingParameter)

spinifex/handlers/iam/service.go

@@ -26,6 +26,7 @@ type IAMService interface {
 	ListPolicyVersions(accountID string, input *iam.ListPolicyVersionsInput) (*iam.ListPolicyVersionsOutput, error)
 	ListPolicies(accountID string, input *iam.ListPoliciesInput) (*iam.ListPoliciesOutput, error)
 	DeletePolicy(accountID string, input *iam.DeletePolicyInput) (*iam.DeletePolicyOutput, error)
+	ListEntitiesForPolicy(accountID string, input *iam.ListEntitiesForPolicyInput) (*iam.ListEntitiesForPolicyOutput, error)
 
 	// Policy attachment — account-scoped
 	AttachUserPolicy(accountID string, input *iam.AttachUserPolicyInput) (*iam.AttachUserPolicyOutput, error)

spinifex/handlers/iam/service_impl.go

@@ -1187,6 +1187,10 @@ func (s *IAMServiceImpl) CreatePolicy(accountID string, input *iam.CreatePolicyI
 			"policy %q: %w", policyName, err)
 	}
 
+	if err := validateTags(input.Tags); err != nil {
+		return nil, err
+	}
+
 	path := aws.StringValue(input.Path)
 	if path == "" {
 		path = "/"
@@ -1207,7 +1211,7 @@ func (s *IAMServiceImpl) CreatePolicy(accountID string, input *iam.CreatePolicyI
 		PolicyDocument: *input.PolicyDocument,
 		CreatedAt:      time.Now().UTC().Format(time.RFC3339),
 		DefaultVersion: "v1",
-		Tags:           []Tag{},
+		Tags:           copyTags(input.Tags),
 	}
 
 	data, err := json.Marshal(policy)
@@ -1236,6 +1240,7 @@ func (s *IAMServiceImpl) CreatePolicy(accountID string, input *iam.CreatePolicyI
 			CreateDate:       aws.Time(createdAt),
 			AttachmentCount:  aws.Int64(0),
 			IsAttachable:     aws.Bool(true),
+			Tags:             tagsToSDK(policy.Tags),
 		},
 	}, nil
 }
@@ -1265,6 +1270,7 @@ func (s *IAMServiceImpl) GetPolicy(accountID string, input *iam.GetPolicyInput)
 			CreateDate:       aws.Time(createdAt),
 			AttachmentCount:  aws.Int64(attachmentCount),
 			IsAttachable:     aws.Bool(true),
+			Tags:             tagsToSDK(policy.Tags),
 		},
 	}, nil
 }
@@ -1365,6 +1371,7 @@ func (s *IAMServiceImpl) ListPolicies(accountID string, input *iam.ListPoliciesI
 			CreateDate:       aws.Time(createdAt),
 			AttachmentCount:  aws.Int64(attachCounts[policy.ARN]),
 			IsAttachable:     aws.Bool(true),
+			Tags:             tagsToSDK(policy.Tags),
 		})
 	}
 
@@ -1398,6 +1405,76 @@ func (s *IAMServiceImpl) DeletePolicy(accountID string, input *iam.DeletePolicyI
 	return &iam.DeletePolicyOutput{}, nil
 }
 
+// ListEntitiesForPolicy reports every user, role, and group that has the given
+// policy attached. The PolicyArn is used as-is against the attachment
+// traversal without first resolving it to a stored Policy record, so an
+// AWS-managed ARN (never provisioned locally, but attachable and stored
+// opaquely on the principal's AttachedPolicies) is found the same way a
+// customer-managed one is.
+func (s *IAMServiceImpl) ListEntitiesForPolicy(accountID string, input *iam.ListEntitiesForPolicyInput) (*iam.ListEntitiesForPolicyOutput, error) {
+	ctx := context.Background()
+	policyARN := *input.PolicyArn
+
+	empty := &iam.ListEntitiesForPolicyOutput{
+		PolicyUsers:  make([]*iam.PolicyUser, 0),
+		PolicyRoles:  make([]*iam.PolicyRole, 0),
+		PolicyGroups: make([]*iam.PolicyGroup, 0),
+		IsTruncated:  aws.Bool(false),
+	}
+
+	// Permissions boundaries are rejected outright at policy-attachment time,
+	// so no entity ever uses a policy that way. The correct answer is empty.
+	if aws.StringValue(input.PolicyUsageFilter) == iam.PolicyUsageTypePermissionsBoundary {
+		return empty, nil
+	}
+
+	entityFilter := aws.StringValue(input.EntityFilter)
+	switch entityFilter {
+	case "", iam.EntityTypeUser, iam.EntityTypeRole, iam.EntityTypeGroup,
+		iam.EntityTypeLocalManagedPolicy, iam.EntityTypeAwsmanagedPolicy:
+		// valid
+	default:
+		return nil, errors.New(awserrors.ErrorIAMInvalidInput)
+	}
+
+	// LocalManagedPolicy/AWSManagedPolicy describe the policy's own management
+	// type, not an entity kind this call returns, so no user/role/group ever
+	// matches one of those two filter values.
+	if entityFilter == iam.EntityTypeLocalManagedPolicy || entityFilter == iam.EntityTypeAwsmanagedPolicy {
+		return empty, nil
+	}
+
+	pathPrefix := aws.StringValue(input.PathPrefix)
+	if pathPrefix == "" {
+		pathPrefix = "/"
+	}
+
+	attachments, err := s.buildPolicyAttachments(ctx, accountID)
+	if err != nil {
+		return nil, fmt.Errorf("list entities for policy: %w", err)
+	}
+
+	out := empty
+	for _, ref := range attachments[policyARN] {
+		if !strings.HasPrefix(ref.path, pathPrefix) {
+			continue
+		}
+		if entityFilter != "" && entityFilter != ref.kind {
+			continue
+		}
+		switch ref.kind {
+		case iam.EntityTypeUser:
+			out.PolicyUsers = append(out.PolicyUsers, &iam.PolicyUser{UserId: aws.String(ref.id), UserName: aws.String(ref.name)})
+		case iam.EntityTypeRole:
+			out.PolicyRoles = append(out.PolicyRoles, &iam.PolicyRole{RoleId: aws.String(ref.id), RoleName: aws.String(ref.name)})
+		case iam.EntityTypeGroup:
+			out.PolicyGroups = append(out.PolicyGroups, &iam.PolicyGroup{GroupId: aws.String(ref.id), GroupName: aws.String(ref.name)})
+		}
+	}
+
+	return out, nil
+}
+
 // ---------------------------------------------------------------------------
 // Policy Attachment
 // ---------------------------------------------------------------------------
@@ -1873,46 +1950,118 @@ func (s *IAMServiceImpl) getPolicyByARN(ctx context.Context, accountID, policyAR
 	return &policy, nil
 }
 
-// buildAttachmentCounts fetches all users for the account once and returns a
-// map of policyARN -> number of users that have it attached, avoiding a
-// per-policy full scan when listing or inspecting policies.
-func (s *IAMServiceImpl) buildAttachmentCounts(ctx context.Context, accountID string) (map[string]int64, error) {
-	keys, err := kvutil.Keys(ctx, s.usersBucket)
-	if err != nil {
-		if errors.Is(err, jetstream.ErrNoKeysFound) {
-			return nil, nil
-		}
-		return nil, fmt.Errorf("build attachment counts: %w", err)
+// entityRef identifies one principal record found to have a given policy
+// attached. It carries enough of the stored record to serve both attachment
+// counts and ListEntitiesForPolicy without a second lookup.
+type entityRef struct {
+	kind string // iam.EntityTypeUser, iam.EntityTypeRole, or iam.EntityTypeGroup
+	name string
+	id   string
+	path string
+}
+
+// attachedPrincipal is the bucket-agnostic shape buildPolicyAttachments needs
+// out of a User, Role, or Group record.
+type attachedPrincipal struct {
+	name     string
+	id       string
+	path     string
+	attached []string
+}
+
+// buildPolicyAttachments scans the users, roles, and groups buckets once and
+// returns, for every attached policy ARN, every principal that attaches it.
+// GetPolicy, ListPolicies, and DeletePolicy only need the count, but
+// ListEntitiesForPolicy needs the identities, so one traversal serves both
+// rather than keeping a count-only copy that would drift from this one. This
+// triples an already O(principals) scan on every GetPolicy, ListPolicies, and
+// DeletePolicy call; acceptable at current scale, but worth knowing about
+// before assuming any of those three got slower for an unrelated reason.
+func (s *IAMServiceImpl) buildPolicyAttachments(ctx context.Context, accountID string) (map[string][]entityRef, error) {
+	attachments := make(map[string][]entityRef)
+
+	scans := []struct {
+		bucket jetstream.KeyValue
+		kind   string
+		decode func([]byte) (attachedPrincipal, error)
+	}{
+		{s.usersBucket, iam.EntityTypeUser, func(data []byte) (attachedPrincipal, error) {
+			var u User
+			err := json.Unmarshal(data, &u)
+			return attachedPrincipal{name: u.UserName, id: u.UserID, path: u.Path, attached: u.AttachedPolicies}, err
+		}},
+		{s.rolesBucket, iam.EntityTypeRole, func(data []byte) (attachedPrincipal, error) {
+			var r Role
+			err := json.Unmarshal(data, &r)
+			return attachedPrincipal{name: r.RoleName, id: r.RoleID, path: r.Path, attached: r.AttachedPolicies}, err
+		}},
+		{s.groupsBucket, iam.EntityTypeGroup, func(data []byte) (attachedPrincipal, error) {
+			var g Group
+			err := json.Unmarshal(data, &g)
+			return attachedPrincipal{name: g.GroupName, id: g.GroupID, path: g.Path, attached: g.AttachedPolicies}, err
+		}},
 	}
 
-	counts := make(map[string]int64)
 	keyPrefix := accountID + "."
-	for _, key := range keys {
-		if key == utils.VersionKey {
-			continue
-		}
-		if !strings.HasPrefix(key, keyPrefix) {
-			continue
-		}
-
-		entry, err := s.usersBucket.Get(ctx, key)
+	for _, scan := range scans {
+		keys, err := kvutil.Keys(ctx, scan.bucket)
 		if err != nil {
-			if errors.Is(err, jetstream.ErrKeyNotFound) {
-				slog.Debug("buildAttachmentCounts: user key disappeared", "key", key)
+			if errors.Is(err, jetstream.ErrNoKeysFound) {
 				continue
 			}
-			slog.Warn("buildAttachmentCounts: failed to get user", "key", key, "err", err)
-			continue
+			return nil, fmt.Errorf("build policy attachments (%s): %w", scan.kind, err)
 		}
-		var user User
-		if err := json.Unmarshal(entry.Value(), &user); err != nil {
-			slog.Warn("buildAttachmentCounts: failed to unmarshal user", "key", key, "err", err)
-			continue
-		}
-		for _, arn := range user.AttachedPolicies {
-			counts[arn]++
+
+		for _, key := range keys {
+			if key == utils.VersionKey {
+				continue
+			}
+			if !strings.HasPrefix(key, keyPrefix) {
+				continue
+			}
+
+			entry, err := scan.bucket.Get(ctx, key)
+			if err != nil {
+				if errors.Is(err, jetstream.ErrKeyNotFound) {
+					slog.Debug("buildPolicyAttachments: key disappeared (concurrent delete)", "kind", scan.kind, "key", key)
+					continue
+				}
+				slog.Warn("buildPolicyAttachments: failed to get record", "kind", scan.kind, "key", key, "err", err)
+				continue
+			}
+
+			principal, err := scan.decode(entry.Value())
+			if err != nil {
+				slog.Warn("buildPolicyAttachments: failed to unmarshal record", "kind", scan.kind, "key", key, "err", err)
+				continue
+			}
+
+			for _, policyARN := range principal.attached {
+				attachments[policyARN] = append(attachments[policyARN], entityRef{
+					kind: scan.kind,
+					name: principal.name,
+					id:   principal.id,
+					path: principal.path,
+				})
+			}
 		}
 	}
+
+	return attachments, nil
+}
+
+// buildAttachmentCounts derives per-policy attachment counts from the shared
+// traversal so GetPolicy, ListPolicies, and DeletePolicy don't carry their own
+// copy of the count-only logic.
+func (s *IAMServiceImpl) buildAttachmentCounts(ctx context.Context, accountID string) (map[string]int64, error) {
+	attachments, err := s.buildPolicyAttachments(ctx, accountID)
+	if err != nil {
+		return nil, err
+	}
+	counts := make(map[string]int64, len(attachments))
+	for policyARN, refs := range attachments {
+		counts[policyARN] = int64(len(refs))
+	}
 	return counts, nil
 }
 

spinifex/handlers/iam/service_impl_test.go

@@ -1206,6 +1206,75 @@ func TestCreatePolicy_ArrayActions(t *testing.T) {
 	assert.Equal(t, "ArrayActions", *out.Policy.PolicyName)
 }
 
+func TestCreatePolicy_WithTags(t *testing.T) {
+	t.Parallel()
+	svc := setupTestIAMService(t)
+
+	tags := []*iam.Tag{
+		{Key: aws.String("team"), Value: aws.String("platform")},
+		{Key: aws.String("env"), Value: aws.String("prod")},
+	}
+	createOut, err := svc.CreatePolicy(testAccountID, &iam.CreatePolicyInput{
+		PolicyName:     aws.String("TaggedPolicy"),
+		PolicyDocument: aws.String(validPolicyDocument()),
+		Tags:           tags,
+	})
+	require.NoError(t, err)
+	require.Len(t, createOut.Policy.Tags, 2)
+
+	getOut, err := svc.GetPolicy(testAccountID, &iam.GetPolicyInput{PolicyArn: createOut.Policy.Arn})
+	require.NoError(t, err)
+	require.Len(t, getOut.Policy.Tags, 2)
+
+	listOut, err := svc.ListPolicies(testAccountID, &iam.ListPoliciesInput{})
+	require.NoError(t, err)
+	var listed *iam.Policy
+	for _, p := range listOut.Policies {
+		if *p.PolicyName == "TaggedPolicy" {
+			listed = p
+		}
+	}
+	require.NotNil(t, listed, "TaggedPolicy missing from ListPolicies")
+	require.Len(t, listed.Tags, 2)
+
+	tagsOut, err := svc.ListPolicyTags(testAccountID, &iam.ListPolicyTagsInput{PolicyArn: createOut.Policy.Arn})
+	require.NoError(t, err)
+	require.Len(t, tagsOut.Tags, 2)
+
+	byKey := func(tags []*iam.Tag) map[string]string {
+		m := make(map[string]string, len(tags))
+		for _, tag := range tags {
+			m[*tag.Key] = aws.StringValue(tag.Value)
+		}
+		return m
+	}
+	want := map[string]string{"team": "platform", "env": "prod"}
+	assert.Equal(t, want, byKey(createOut.Policy.Tags))
+	assert.Equal(t, want, byKey(getOut.Policy.Tags))
+	assert.Equal(t, want, byKey(listed.Tags))
+	assert.Equal(t, want, byKey(tagsOut.Tags))
+}
+
+func TestCreatePolicy_InvalidTag(t *testing.T) {
+	t.Parallel()
+	svc := setupTestIAMService(t)
+
+	_, err := svc.CreatePolicy(testAccountID, &iam.CreatePolicyInput{
+		PolicyName:     aws.String("BadTagPolicy"),
+		PolicyDocument: aws.String(validPolicyDocument()),
+		Tags:           []*iam.Tag{{Key: aws.String(""), Value: aws.String("x")}},
+	})
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), awserrors.ErrorIAMInvalidInput)
+
+	// The rejected policy must not have been stored.
+	_, err = svc.GetPolicy(testAccountID, &iam.GetPolicyInput{
+		PolicyArn: aws.String("arn:aws:iam::" + testAccountID + ":policy/BadTagPolicy"),
+	})
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), awserrors.ErrorIAMNoSuchEntity)
+}
+
 func TestGetPolicy(t *testing.T) {
 	t.Parallel()
 	svc := setupTestIAMService(t)
@@ -1470,6 +1539,242 @@ func TestDeletePolicy_AttachedConflict(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestGetPolicy_AttachmentCount_AllPrincipalKinds(t *testing.T) {
+	t.Parallel()
+	svc := setupTestIAMService(t)
+
+	userPolicy := createTestPolicy(t, svc, "UserAttached")
+	rolePolicy := createTestPolicy(t, svc, "RoleAttached")
+	groupPolicy := createTestPolicy(t, svc, "GroupAttached")
+
+	createTestUser(t, svc, "countuser")
+	createTestRole(t, svc, "countrole")
+	createTestGroup(t, svc, "countgroup")
+
+	_, err := svc.AttachUserPolicy(testAccountID, &iam.AttachUserPolicyInput{
+		UserName: aws.String("countuser"), PolicyArn: userPolicy.Arn,
+	})
+	require.NoError(t, err)
+	_, err = svc.AttachRolePolicy(testAccountID, &iam.AttachRolePolicyInput{
+		RoleName: aws.String("countrole"), PolicyArn: rolePolicy.Arn,
+	})
+	require.NoError(t, err)
+	_, err = svc.AttachGroupPolicy(testAccountID, &iam.AttachGroupPolicyInput{
+		GroupName: aws.String("countgroup"), PolicyArn: groupPolicy.Arn,
+	})
+	require.NoError(t, err)
+
+	for _, tc := range []struct {
+		name string
+		arn  *string
+	}{
+		{"user", userPolicy.Arn},
+		{"role", rolePolicy.Arn},
+		{"group", groupPolicy.Arn},
+	} {
+		out, err := svc.GetPolicy(testAccountID, &iam.GetPolicyInput{PolicyArn: tc.arn})
+		require.NoError(t, err)
+		assert.Equal(t, int64(1), *out.Policy.AttachmentCount, "%s-attached policy", tc.name)
+	}
+}
+
+func TestDeletePolicy_AttachedConflict_Role(t *testing.T) {
+	t.Parallel()
+	svc := setupTestIAMService(t)
+	created := createTestPolicy(t, svc, "RoleConflict")
+	createTestRole(t, svc, "conflictrole")
+
+	_, err := svc.AttachRolePolicy(testAccountID, &iam.AttachRolePolicyInput{
+		RoleName:  aws.String("conflictrole"),
+		PolicyArn: created.Arn,
+	})
+	require.NoError(t, err)
+
+	_, err = svc.DeletePolicy(testAccountID, &iam.DeletePolicyInput{PolicyArn: created.Arn})
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), awserrors.ErrorIAMDeleteConflict)
+
+	_, err = svc.DetachRolePolicy(testAccountID, &iam.DetachRolePolicyInput{
+		RoleName:  aws.String("conflictrole"),
+		PolicyArn: created.Arn,
+	})
+	require.NoError(t, err)
+
+	_, err = svc.DeletePolicy(testAccountID, &iam.DeletePolicyInput{PolicyArn: created.Arn})
+	require.NoError(t, err)
+}
+
+func TestDeletePolicy_AttachedConflict_Group(t *testing.T) {
+	t.Parallel()
+	svc := setupTestIAMService(t)
+	created := createTestPolicy(t, svc, "GroupConflict")
+	createTestGroup(t, svc, "conflictgroup")
+
+	_, err := svc.AttachGroupPolicy(testAccountID, &iam.AttachGroupPolicyInput{
+		GroupName: aws.String("conflictgroup"),
+		PolicyArn: created.Arn,
+	})
+	require.NoError(t, err)
+
+	_, err = svc.DeletePolicy(testAccountID, &iam.DeletePolicyInput{PolicyArn: created.Arn})
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), awserrors.ErrorIAMDeleteConflict)
+
+	_, err = svc.DetachGroupPolicy(testAccountID, &iam.DetachGroupPolicyInput{
+		GroupName: aws.String("conflictgroup"),
+		PolicyArn: created.Arn,
+	})
+	require.NoError(t, err)
+
+	_, err = svc.DeletePolicy(testAccountID, &iam.DeletePolicyInput{PolicyArn: created.Arn})
+	require.NoError(t, err)
+}
+
+// ============================================================================
+// ListEntitiesForPolicy Tests
+// ============================================================================
+
+func TestListEntitiesForPolicy_Empty(t *testing.T) {
+	t.Parallel()
+	svc := setupTestIAMService(t)
+	created := createTestPolicy(t, svc, "NoEntities")
+
+	out, err := svc.ListEntitiesForPolicy(testAccountID, &iam.ListEntitiesForPolicyInput{
+		PolicyArn: created.Arn,
+	})
+	require.NoError(t, err)
+	require.NotNil(t, out.PolicyUsers, emptyRequiredListMsg("ListEntitiesForPolicy", "PolicyUsers"))
+	require.NotNil(t, out.PolicyRoles, emptyRequiredListMsg("ListEntitiesForPolicy", "PolicyRoles"))
+	require.NotNil(t, out.PolicyGroups, emptyRequiredListMsg("ListEntitiesForPolicy", "PolicyGroups"))
+	assert.Empty(t, out.PolicyUsers)
+	assert.Empty(t, out.PolicyRoles)
+	assert.Empty(t, out.PolicyGroups)
+}
+
+func TestListEntitiesForPolicy_AllKinds(t *testing.T) {
+	t.Parallel()
+	svc := setupTestIAMService(t)
+	created := createTestPolicy(t, svc, "Shared")
+
+	createTestUser(t, svc, "entuser")
+	createTestRole(t, svc, "entrole")
+	createTestGroup(t, svc, "entgroup")
+
+	_, err := svc.AttachUserPolicy(testAccountID, &iam.AttachUserPolicyInput{UserName: aws.String("entuser"), PolicyArn: created.Arn})
+	require.NoError(t, err)
+	_, err = svc.AttachRolePolicy(testAccountID, &iam.AttachRolePolicyInput{RoleName: aws.String("entrole"), PolicyArn: created.Arn})
+	require.NoError(t, err)
+	_, err = svc.AttachGroupPolicy(testAccountID, &iam.AttachGroupPolicyInput{GroupName: aws.String("entgroup"), PolicyArn: created.Arn})
+	require.NoError(t, err)
+
+	out, err := svc.ListEntitiesForPolicy(testAccountID, &iam.ListEntitiesForPolicyInput{PolicyArn: created.Arn})
+	require.NoError(t, err)
+	require.Len(t, out.PolicyUsers, 1)
+	require.Len(t, out.PolicyRoles, 1)
+	require.Len(t, out.PolicyGroups, 1)
+	assert.Equal(t, "entuser", *out.PolicyUsers[0].UserName)
+	assert.Equal(t, "entrole", *out.PolicyRoles[0].RoleName)
+	assert.Equal(t, "entgroup", *out.PolicyGroups[0].GroupName)
+}
+
+func TestListEntitiesForPolicy_EntityFilter(t *testing.T) {
+	t.Parallel()
+	svc := setupTestIAMService(t)
+	created := createTestPolicy(t, svc, "FilterMe")
+
+	createTestUser(t, svc, "filteruser")
+	createTestRole(t, svc, "filterrole")
+	createTestGroup(t, svc, "filtergroup")
+
+	_, err := svc.AttachUserPolicy(testAccountID, &iam.AttachUserPolicyInput{UserName: aws.String("filteruser"), PolicyArn: created.Arn})
+	require.NoError(t, err)
+	_, err = svc.AttachRolePolicy(testAccountID, &iam.AttachRolePolicyInput{RoleName: aws.String("filterrole"), PolicyArn: created.Arn})
+	require.NoError(t, err)
+	_, err = svc.AttachGroupPolicy(testAccountID, &iam.AttachGroupPolicyInput{GroupName: aws.String("filtergroup"), PolicyArn: created.Arn})
+	require.NoError(t, err)
+
+	out, err := svc.ListEntitiesForPolicy(testAccountID, &iam.ListEntitiesForPolicyInput{
+		PolicyArn:    created.Arn,
+		EntityFilter: aws.String(iam.EntityTypeRole),
+	})
+	require.NoError(t, err)
+	assert.Empty(t, out.PolicyUsers)
+	assert.Empty(t, out.PolicyGroups)
+	require.Len(t, out.PolicyRoles, 1)
+	assert.Equal(t, "filterrole", *out.PolicyRoles[0].RoleName)
+}
+
+func TestListEntitiesForPolicy_PolicyTypeFilterYieldsNoEntities(t *testing.T) {
+	t.Parallel()
+	svc := setupTestIAMService(t)
+	created := createTestPolicy(t, svc, "PolicyTypeFilter")
+	createTestUser(t, svc, "ptfuser")
+	_, err := svc.AttachUserPolicy(testAccountID, &iam.AttachUserPolicyInput{UserName: aws.String("ptfuser"), PolicyArn: created.Arn})
+	require.NoError(t, err)
+
+	// LocalManagedPolicy/AWSManagedPolicy describe the policy's own management
+	// type; ListEntitiesForPolicy only ever returns users/roles/groups, so
+	// neither value can match any collected entity.
+	out, err := svc.ListEntitiesForPolicy(testAccountID, &iam.ListEntitiesForPolicyInput{
+		PolicyArn:    created.Arn,
+		EntityFilter: aws.String(iam.EntityTypeLocalManagedPolicy),
+	})
+	require.NoError(t, err)
+	assert.Empty(t, out.PolicyUsers)
+	assert.Empty(t, out.PolicyRoles)
+	assert.Empty(t, out.PolicyGroups)
+}
+
+func TestListEntitiesForPolicy_PathPrefix(t *testing.T) {
+	t.Parallel()
+	svc := setupTestIAMService(t)
+	created := createTestPolicy(t, svc, "PathFiltered")
+
+	_, err := svc.CreateUser(testAccountID, &iam.CreateUserInput{
+		UserName: aws.String("devuser"), Path: aws.String("/dev/"),
+	})
+	require.NoError(t, err)
+	_, err = svc.CreateUser(testAccountID, &iam.CreateUserInput{
+		UserName: aws.String("opsuser"), Path: aws.String("/ops/"),
+	})
+	require.NoError(t, err)
+
+	_, err = svc.AttachUserPolicy(testAccountID, &iam.AttachUserPolicyInput{UserName: aws.String("devuser"), PolicyArn: created.Arn})
+	require.NoError(t, err)
+	_, err = svc.AttachUserPolicy(testAccountID, &iam.AttachUserPolicyInput{UserName: aws.String("opsuser"), PolicyArn: created.Arn})
+	require.NoError(t, err)
+
+	out, err := svc.ListEntitiesForPolicy(testAccountID, &iam.ListEntitiesForPolicyInput{
+		PolicyArn:  created.Arn,
+		PathPrefix: aws.String("/dev/"),
+	})
+	require.NoError(t, err)
+	require.Len(t, out.PolicyUsers, 1)
+	assert.Equal(t, "devuser", *out.PolicyUsers[0].UserName)
+}
+
+func TestListEntitiesForPolicy_AWSManagedARN(t *testing.T) {
+	t.Parallel()
+	svc := setupTestIAMService(t)
+	managedARN := "arn:aws:iam::aws:policy/AmazonEKSClusterPolicy"
+	createTestRole(t, svc, "eksrole")
+
+	_, err := svc.AttachRolePolicy(testAccountID, &iam.AttachRolePolicyInput{
+		RoleName: aws.String("eksrole"), PolicyArn: aws.String(managedARN),
+	})
+	require.NoError(t, err)
+
+	// AWS-managed ARNs are stored opaquely and never appear in the policies
+	// bucket, so this call resolves entities purely from the attachment
+	// traversal rather than a lookup keyed on a stored Policy record.
+	out, err := svc.ListEntitiesForPolicy(testAccountID, &iam.ListEntitiesForPolicyInput{
+		PolicyArn: aws.String(managedARN),
+	})
+	require.NoError(t, err)
+	require.Len(t, out.PolicyRoles, 1)
+	assert.Equal(t, "eksrole", *out.PolicyRoles[0].RoleName)
+}
+
 // ============================================================================
 // Policy Attachment Tests
 // ============================================================================

tests/integration/cross_tenant_test.go

@@ -473,10 +473,10 @@ func elbv2Uncased() map[string]string {
 
 func iamUncased() map[string]string {
 	uncased := map[string]string{}
-	// The six siblings of GetPolicy read PolicyArn through the same branch.
+	// The seven siblings of GetPolicy read PolicyArn through the same branch.
 	for _, action := range []string{
 		"GetPolicyVersion", "ListPolicyVersions", "DeletePolicy", "TagPolicy", "UntagPolicy",
-		"ListPolicyTags",
+		"ListPolicyTags", "ListEntitiesForPolicy",
 	} {
 		uncased[action] = sameHelper("GetPolicy")
 	}