diff --git a/cmd/list-app-owners.go b/cmd/list-app-owners.go
index eba29df..59a511d 100644
--- a/cmd/list-app-owners.go
+++ b/cmd/list-app-owners.go
@@ -91,11 +91,11 @@ func listAppOwners(ctx context.Context, client client.AzureClient, apps <-chan i
 			for id := range stream {
 				var (
 					data = models.AppOwners{
-						AppId: id.(string),
+						AppId: id,
 					}
 					count = 0
 				)
-				for item := range client.ListAzureADAppOwners(ctx, id.(string), "", "", "", nil) {
+				for item := range client.ListAzureADAppOwners(ctx, id, "", "", "", nil) {
 					if item.Error != nil {
 						log.Error(item.Error, "unable to continue processing owners for this app", "appId", id)
 					} else {
diff --git a/cmd/list-azure-rm.go b/cmd/list-azure-rm.go
index 0f9cc20..74fbb69 100644
--- a/cmd/list-azure-rm.go
+++ b/cmd/list-azure-rm.go
@@ -26,6 +26,7 @@ import (
 
 	"github.com/bloodhoundad/azurehound/client"
 	"github.com/bloodhoundad/azurehound/enums"
+	"github.com/bloodhoundad/azurehound/models"
 	"github.com/bloodhoundad/azurehound/pipeline"
 	"github.com/spf13/cobra"
 )
@@ -70,7 +71,11 @@ func listAllRM(ctx context.Context, client client.AzureClient) <-chan interface{
 		keyVaults  = make(chan interface{})
 		keyVaults2 = make(chan interface{})
 		keyVaults3 = make(chan interface{})
-		keyVaults4 = make(chan interface{})
+
+		keyVaultRoleAssignments1 = make(chan azureWrapper[models.KeyVaultRoleAssignments])
+		keyVaultRoleAssignments2 = make(chan azureWrapper[models.KeyVaultRoleAssignments])
+		keyVaultRoleAssignments3 = make(chan azureWrapper[models.KeyVaultRoleAssignments])
+		keyVaultRoleAssignments4 = make(chan azureWrapper[models.KeyVaultRoleAssignments])
 
 		mgmtGroups  = make(chan interface{})
 		mgmtGroups2 = make(chan interface{})
@@ -102,10 +107,14 @@ func listAllRM(ctx context.Context, client client.AzureClient) <-chan interface{
 	subscriptionUserAccessAdmins := listSubscriptionUserAccessAdmins(ctx, client, subscriptions6)
 
 	// Enumerate KeyVaults, KeyVaultOwners, KeyVaultAccessPolicies and KeyVaultUserAccessAdmins
-	pipeline.Tee(ctx.Done(), listKeyVaults(ctx, client, subscriptions2), keyVaults, keyVaults2, keyVaults3, keyVaults4)
-	keyVaultOwners := listKeyVaultOwners(ctx, client, keyVaults2)
+	pipeline.Tee(ctx.Done(), listKeyVaults(ctx, client, subscriptions2), keyVaults, keyVaults2, keyVaults3)
+	pipeline.Tee(ctx.Done(), listKeyVaultRoleAssignments(ctx, client, keyVaults2), keyVaultRoleAssignments1, keyVaultRoleAssignments2, keyVaultRoleAssignments3, keyVaultRoleAssignments4)
 	keyVaultAccessPolicies := listKeyVaultAccessPolicies(ctx, client, keyVaults3, []enums.KeyVaultAccessType{enums.GetCerts, enums.GetKeys, enums.GetCerts})
-	keyVaultUserAccessAdmins := listKeyVaultUserAccessAdmins(ctx, client, keyVaults4)
+
+	keyVaultOwners := listKeyVaultOwners(ctx, keyVaultRoleAssignments1)
+	keyVaultUserAccessAdmins := listKeyVaultUserAccessAdmins(ctx, keyVaultRoleAssignments2)
+	keyVaultContributors := listKeyVaultContributors(ctx, keyVaultRoleAssignments3)
+	keyVaultKVContributors := listKeyVaultKVContributors(ctx, keyVaultRoleAssignments4)
 
 	// Enumerate ManagementGroups, ManagementGroupOwners and ManagementGroupDescendants
 	pipeline.Tee(ctx.Done(), listManagementGroups(ctx, client), mgmtGroups, mgmtGroups2, mgmtGroups3, mgmtGroups4)
@@ -129,6 +138,8 @@ func listAllRM(ctx context.Context, client client.AzureClient) <-chan interface{
 
 	return pipeline.Mux(ctx.Done(),
 		keyVaultAccessPolicies,
+		keyVaultContributors,
+		keyVaultKVContributors,
 		keyVaultOwners,
 		keyVaultUserAccessAdmins,
 		keyVaults,
diff --git a/cmd/list-device-owners.go b/cmd/list-device-owners.go
index 7950668..a60ccde 100644
--- a/cmd/list-device-owners.go
+++ b/cmd/list-device-owners.go
@@ -90,11 +90,11 @@ func listDeviceOwners(ctx context.Context, client client.AzureClient, devices <-
 			for id := range stream {
 				var (
 					data = models.DeviceOwners{
-						DeviceId: id.(string),
+						DeviceId: id,
 					}
 					count = 0
 				)
-				for item := range client.ListAzureDeviceRegisteredOwners(ctx, id.(string), false) {
+				for item := range client.ListAzureDeviceRegisteredOwners(ctx, id, false) {
 					if item.Error != nil {
 						log.Error(item.Error, "unable to continue processing owners for this device", "deviceId", id)
 					} else {
diff --git a/cmd/list-group-members.go b/cmd/list-group-members.go
index dc51f93..51e9478 100644
--- a/cmd/list-group-members.go
+++ b/cmd/list-group-members.go
@@ -91,11 +91,11 @@ func listGroupMembers(ctx context.Context, client client.AzureClient, groups <-c
 			for id := range stream {
 				var (
 					data = models.GroupMembers{
-						GroupId: id.(string),
+						GroupId: id,
 					}
 					count = 0
 				)
-				for item := range client.ListAzureADGroupMembers(ctx, id.(string), "", "", "", nil) {
+				for item := range client.ListAzureADGroupMembers(ctx, id, "", "", "", nil) {
 					if item.Error != nil {
 						log.Error(item.Error, "unable to continue processing members for this group", "groupId", id)
 					} else {
diff --git a/cmd/list-group-owners.go b/cmd/list-group-owners.go
index 01141de..7cfb051 100644
--- a/cmd/list-group-owners.go
+++ b/cmd/list-group-owners.go
@@ -91,11 +91,11 @@ func listGroupOwners(ctx context.Context, client client.AzureClient, groups <-ch
 			for id := range stream {
 				var (
 					groupOwners = models.GroupOwners{
-						GroupId: id.(string),
+						GroupId: id,
 					}
 					count = 0
 				)
-				for item := range client.ListAzureADGroupOwners(ctx, id.(string), "", "", "", nil) {
+				for item := range client.ListAzureADGroupOwners(ctx, id, "", "", "", nil) {
 					if item.Error != nil {
 						log.Error(item.Error, "unable to continue processing owners for this group", "groupId", id)
 					} else {
diff --git a/cmd/list-key-vault-contributors.go b/cmd/list-key-vault-contributors.go
index 77d8032..1e017e7 100644
--- a/cmd/list-key-vault-contributors.go
+++ b/cmd/list-key-vault-contributors.go
@@ -19,15 +19,13 @@ package cmd
 
 import (
 	"context"
-	"fmt"
 	"os"
 	"os/signal"
-	"path"
 	"time"
 
-	"github.com/bloodhoundad/azurehound/client"
 	"github.com/bloodhoundad/azurehound/constants"
 	"github.com/bloodhoundad/azurehound/enums"
+	"github.com/bloodhoundad/azurehound/internal"
 	"github.com/bloodhoundad/azurehound/models"
 	"github.com/bloodhoundad/azurehound/pipeline"
 	"github.com/spf13/cobra"
@@ -59,52 +57,30 @@ func listKeyVaultContributorsCmdImpl(cmd *cobra.Command, args []string) {
 		subscriptions := listSubscriptions(ctx, azClient)
 		keyVaults := listKeyVaults(ctx, azClient, subscriptions)
 		kvRoleAssignments := listKeyVaultRoleAssignments(ctx, azClient, keyVaults)
-		stream := listKeyVaultContributors(ctx, azClient, kvRoleAssignments)
+		stream := listKeyVaultContributors(ctx, kvRoleAssignments)
 		outputStream(ctx, stream)
 		duration := time.Since(start)
 		log.Info("collection completed", "duration", duration.String())
 	}
 }
 
-func listKeyVaultContributors(ctx context.Context, client client.AzureClient, vmRoleAssignments <-chan interface{}) <-chan interface{} {
-	out := make(chan interface{})
+func listKeyVaultContributors(
+	ctx context.Context,
+	kvRoleAssignments <-chan azureWrapper[models.KeyVaultRoleAssignments],
+) <-chan any {
+	return pipeline.Map(ctx.Done(), kvRoleAssignments, func(ra azureWrapper[models.KeyVaultRoleAssignments]) any {
+		filteredAssignments := internal.Filter(ra.Data.RoleAssignments, kvRoleAssignmentFilter(constants.ContributorRoleID))
 
-	go func() {
-		defer close(out)
-
-		for result := range pipeline.OrDone(ctx.Done(), vmRoleAssignments) {
-			if roleAssignments, ok := result.(AzureWrapper).Data.(models.KeyVaultRoleAssignments); !ok {
-				log.Error(fmt.Errorf("failed type assertion"), "unable to continue enumerating key vault contributors", "result", result)
-				return
-			} else {
-				var (
-					keyVaultContributors = models.KeyVaultContributors{
-						KeyVaultId: roleAssignments.KeyVaultId,
-					}
-					count = 0
-				)
-				for _, item := range roleAssignments.RoleAssignments {
-					roleDefinitionId := path.Base(item.RoleAssignment.Properties.RoleDefinitionId)
-
-					if roleDefinitionId == constants.ContributorRoleID {
-						keyVaultContributor := models.KeyVaultContributor{
-							Contributor: item.RoleAssignment,
-							KeyVaultId:  item.KeyVaultId,
-						}
-						log.V(2).Info("found key vault contributor", "keyVaultContributor", keyVaultContributor)
-						count++
-						keyVaultContributors.Contributors = append(keyVaultContributors.Contributors, keyVaultContributor)
-					}
-				}
-				out <- AzureWrapper{
-					Kind: enums.KindAZVMContributor,
-					Data: keyVaultContributors,
-				}
-				log.V(1).Info("finished listing key vault contributors", "keyVaultId", roleAssignments.KeyVaultId, "count", count)
+		contributors := internal.Map(filteredAssignments, func(ra models.KeyVaultRoleAssignment) models.KeyVaultContributor {
+			return models.KeyVaultContributor{
+				ra.RoleAssignment,
+				ra.KeyVaultId,
 			}
-		}
-		log.Info("finished listing all key vault contributors")
-	}()
+		})
 
-	return out
+		return NewAzureWrapper(enums.KindAZKeyVaultContributor, models.KeyVaultContributors{
+			KeyVaultId:   ra.Data.KeyVaultId,
+			Contributors: contributors,
+		})
+	})
 }
diff --git a/cmd/list-key-vault-contributors_test.go b/cmd/list-key-vault-contributors_test.go
index 28efe99..064e9e7 100644
--- a/cmd/list-key-vault-contributors_test.go
+++ b/cmd/list-key-vault-contributors_test.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/bloodhoundad/azurehound/client/mocks"
 	"github.com/bloodhoundad/azurehound/constants"
+	"github.com/bloodhoundad/azurehound/enums"
 	"github.com/bloodhoundad/azurehound/models"
 	"github.com/bloodhoundad/azurehound/models/azure"
 	"github.com/golang/mock/gomock"
@@ -39,7 +40,7 @@ func TestListKeyVaultContributors(t *testing.T) {
 
 	mockClient := mocks.NewMockAzureClient(ctrl)
 
-	mockRoleAssignmentsChannel := make(chan interface{})
+	mockRoleAssignmentsChannel := make(chan azureWrapper[models.KeyVaultRoleAssignments])
 	mockTenant := azure.Tenant{}
 	mockClient.EXPECT().TenantInfo().Return(mockTenant).AnyTimes()
 	channel := listKeyVaultContributors(ctx, mockClient, mockRoleAssignmentsChannel)
@@ -47,29 +48,23 @@ func TestListKeyVaultContributors(t *testing.T) {
 	go func() {
 		defer close(mockRoleAssignmentsChannel)
 
-		mockRoleAssignmentsChannel <- AzureWrapper{
-			Data: models.KeyVaultRoleAssignments{
-				KeyVaultId: "foo",
-				RoleAssignments: []models.KeyVaultRoleAssignment{
-					{
-						RoleAssignment: azure.RoleAssignment{
-							Name: constants.ContributorRoleID,
-							Properties: azure.RoleAssignmentPropertiesWithScope{
-								RoleDefinitionId: constants.ContributorRoleID,
-							},
+		mockRoleAssignmentsChannel <- NewAzureWrapper(enums.KindAZKeyVaultRoleAssignment, models.KeyVaultRoleAssignments{
+			KeyVaultId: "foo",
+			RoleAssignments: []models.KeyVaultRoleAssignment{
+				{
+					RoleAssignment: azure.RoleAssignment{
+						Name: constants.ContributorRoleID,
+						Properties: azure.RoleAssignmentPropertiesWithScope{
+							RoleDefinitionId: constants.ContributorRoleID,
 						},
 					},
 				},
 			},
-		}
+		})
 	}()
 
-	if result, ok := <-channel; !ok {
+	if _, ok := <-channel; !ok {
 		t.Fatalf("failed to receive from channel")
-	} else if wrapper, ok := result.(AzureWrapper); !ok {
-		t.Errorf("failed type assertion: got %T, want %T", result, AzureWrapper{})
-	} else if _, ok := wrapper.Data.(models.KeyVaultContributors); !ok {
-		t.Errorf("failed type assertion: got %T, want %T", wrapper.Data, models.KeyVaultContributors{})
 	}
 
 	if _, ok := <-channel; ok {
diff --git a/cmd/list-key-vault-kvcontributors.go b/cmd/list-key-vault-kvcontributors.go
index d14287c..a9bd4f0 100644
--- a/cmd/list-key-vault-kvcontributors.go
+++ b/cmd/list-key-vault-kvcontributors.go
@@ -19,15 +19,13 @@ package cmd
 
 import (
 	"context"
-	"fmt"
 	"os"
 	"os/signal"
-	"path"
 	"time"
 
-	"github.com/bloodhoundad/azurehound/client"
 	"github.com/bloodhoundad/azurehound/constants"
 	"github.com/bloodhoundad/azurehound/enums"
+	"github.com/bloodhoundad/azurehound/internal"
 	"github.com/bloodhoundad/azurehound/models"
 	"github.com/bloodhoundad/azurehound/pipeline"
 	"github.com/spf13/cobra"
@@ -59,52 +57,30 @@ func listKeyVaultKVContributorsCmdImpl(cmd *cobra.Command, args []string) {
 		subscriptions := listSubscriptions(ctx, azClient)
 		keyVaults := listKeyVaults(ctx, azClient, subscriptions)
 		kvRoleAssignments := listKeyVaultRoleAssignments(ctx, azClient, keyVaults)
-		stream := listKeyVaultKVContributors(ctx, azClient, kvRoleAssignments)
+		stream := listKeyVaultKVContributors(ctx, kvRoleAssignments)
 		outputStream(ctx, stream)
 		duration := time.Since(start)
 		log.Info("collection completed", "duration", duration.String())
 	}
 }
 
-func listKeyVaultKVContributors(ctx context.Context, client client.AzureClient, vmRoleAssignments <-chan interface{}) <-chan interface{} {
-	out := make(chan interface{})
+func listKeyVaultKVContributors(
+	ctx context.Context,
+	kvRoleAssignments <-chan azureWrapper[models.KeyVaultRoleAssignments],
+) <-chan any {
+	return pipeline.Map(ctx.Done(), kvRoleAssignments, func(ra azureWrapper[models.KeyVaultRoleAssignments]) any {
+		filteredAssignments := internal.Filter(ra.Data.RoleAssignments, kvRoleAssignmentFilter(constants.KeyVaultContributorRoleID))
 
-	go func() {
-		defer close(out)
-
-		for result := range pipeline.OrDone(ctx.Done(), vmRoleAssignments) {
-			if roleAssignments, ok := result.(AzureWrapper).Data.(models.KeyVaultRoleAssignments); !ok {
-				log.Error(fmt.Errorf("failed type assertion"), "unable to continue enumerating key vault kvContributors", "result", result)
-				return
-			} else {
-				var (
-					keyVaultKVContributors = models.KeyVaultKVContributors{
-						KeyVaultId: roleAssignments.KeyVaultId,
-					}
-					count = 0
-				)
-				for _, item := range roleAssignments.RoleAssignments {
-					roleDefinitionId := path.Base(item.RoleAssignment.Properties.RoleDefinitionId)
-
-					if roleDefinitionId == constants.KeyVaultContributorRoleID {
-						keyVaultKVContributor := models.KeyVaultKVContributor{
-							KVContributor: item.RoleAssignment,
-							KeyVaultId:    item.KeyVaultId,
-						}
-						log.V(2).Info("found key vault kvContributor", "keyVaultKVContributor", keyVaultKVContributor)
-						count++
-						keyVaultKVContributors.KVContributors = append(keyVaultKVContributors.KVContributors, keyVaultKVContributor)
-					}
-				}
-				out <- AzureWrapper{
-					Kind: enums.KindAZKeyVaultContributor,
-					Data: keyVaultKVContributors,
-				}
-				log.V(1).Info("finished listing key vault kvContributors", "keyVaultId", roleAssignments.KeyVaultId, "count", count)
+		kvContributors := internal.Map(filteredAssignments, func(ra models.KeyVaultRoleAssignment) models.KeyVaultKVContributor {
+			return models.KeyVaultKVContributor{
+				KVContributor: ra.RoleAssignment,
+				KeyVaultId:    ra.KeyVaultId,
 			}
-		}
-		log.Info("finished listing all key vault kvContributors")
-	}()
+		})
 
-	return out
+		return NewAzureWrapper(enums.KindAZKeyVaultKVContributor, models.KeyVaultKVContributors{
+			KeyVaultId:     ra.Data.KeyVaultId,
+			KVContributors: kvContributors,
+		})
+	})
 }
diff --git a/cmd/list-key-vault-kvcontributors_test.go b/cmd/list-key-vault-kvcontributors_test.go
index bc3dc17..63c1193 100644
--- a/cmd/list-key-vault-kvcontributors_test.go
+++ b/cmd/list-key-vault-kvcontributors_test.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/bloodhoundad/azurehound/client/mocks"
 	"github.com/bloodhoundad/azurehound/constants"
+	"github.com/bloodhoundad/azurehound/enums"
 	"github.com/bloodhoundad/azurehound/models"
 	"github.com/bloodhoundad/azurehound/models/azure"
 	"github.com/golang/mock/gomock"
@@ -39,16 +40,17 @@ func TestListKeyVaultKVContributors(t *testing.T) {
 
 	mockClient := mocks.NewMockAzureClient(ctrl)
 
-	mockRoleAssignmentsChannel := make(chan interface{})
+	mockRoleAssignmentsChannel := make(chan azureWrapper[models.KeyVaultRoleAssignments])
 	mockTenant := azure.Tenant{}
 	mockClient.EXPECT().TenantInfo().Return(mockTenant).AnyTimes()
-	channel := listKeyVaultKVContributors(ctx, mockClient, mockRoleAssignmentsChannel)
+	channel := listKeyVaultKVContributors(ctx, mockRoleAssignmentsChannel)
 
 	go func() {
 		defer close(mockRoleAssignmentsChannel)
 
-		mockRoleAssignmentsChannel <- AzureWrapper{
-			Data: models.KeyVaultRoleAssignments{
+		mockRoleAssignmentsChannel <- NewAzureWrapper(
+			enums.KindAZKeyVaultRoleAssignment,
+			models.KeyVaultRoleAssignments{
 				KeyVaultId: "foo",
 				RoleAssignments: []models.KeyVaultRoleAssignment{
 					{
@@ -61,15 +63,11 @@ func TestListKeyVaultKVContributors(t *testing.T) {
 					},
 				},
 			},
-		}
+		)
 	}()
 
-	if result, ok := <-channel; !ok {
+	if _, ok := <-channel; !ok {
 		t.Fatalf("failed to receive from channel")
-	} else if wrapper, ok := result.(AzureWrapper); !ok {
-		t.Errorf("failed type assertion: got %T, want %T", result, AzureWrapper{})
-	} else if _, ok := wrapper.Data.(models.KeyVaultKVContributors); !ok {
-		t.Errorf("failed type assertion: got %T, want %T", wrapper.Data, models.KeyVaultKVContributors{})
 	}
 
 	if _, ok := <-channel; ok {
diff --git a/cmd/list-key-vault-owners.go b/cmd/list-key-vault-owners.go
index ecf954d..a119c98 100644
--- a/cmd/list-key-vault-owners.go
+++ b/cmd/list-key-vault-owners.go
@@ -19,15 +19,13 @@ package cmd
 
 import (
 	"context"
-	"fmt"
 	"os"
 	"os/signal"
-	"path"
 	"time"
 
-	"github.com/bloodhoundad/azurehound/client"
 	"github.com/bloodhoundad/azurehound/constants"
 	"github.com/bloodhoundad/azurehound/enums"
+	"github.com/bloodhoundad/azurehound/internal"
 	"github.com/bloodhoundad/azurehound/models"
 	"github.com/bloodhoundad/azurehound/pipeline"
 	"github.com/spf13/cobra"
@@ -59,52 +57,30 @@ func listKeyVaultOwnersCmdImpl(cmd *cobra.Command, args []string) {
 		subscriptions := listSubscriptions(ctx, azClient)
 		keyVaults := listKeyVaults(ctx, azClient, subscriptions)
 		kvRoleAssignments := listKeyVaultRoleAssignments(ctx, azClient, keyVaults)
-		stream := listKeyVaultOwners(ctx, azClient, kvRoleAssignments)
+		stream := listKeyVaultOwners(ctx, kvRoleAssignments)
 		outputStream(ctx, stream)
 		duration := time.Since(start)
 		log.Info("collection completed", "duration", duration.String())
 	}
 }
 
-func listKeyVaultOwners(ctx context.Context, client client.AzureClient, vmRoleAssignments <-chan interface{}) <-chan interface{} {
-	out := make(chan interface{})
+func listKeyVaultOwners(
+	ctx context.Context,
+	kvRoleAssignments <-chan azureWrapper[models.KeyVaultRoleAssignments],
+) <-chan any {
+	return pipeline.Map(ctx.Done(), kvRoleAssignments, func(ra azureWrapper[models.KeyVaultRoleAssignments]) any {
+		filteredAssignments := internal.Filter(ra.Data.RoleAssignments, kvRoleAssignmentFilter(constants.OwnerRoleID))
 
-	go func() {
-		defer close(out)
-
-		for result := range pipeline.OrDone(ctx.Done(), vmRoleAssignments) {
-			if roleAssignments, ok := result.(AzureWrapper).Data.(models.KeyVaultRoleAssignments); !ok {
-				log.Error(fmt.Errorf("failed type assertion"), "unable to continue enumerating key vault owners", "result", result)
-				return
-			} else {
-				var (
-					keyVaultOwners = models.KeyVaultOwners{
-						KeyVaultId: roleAssignments.KeyVaultId,
-					}
-					count = 0
-				)
-				for _, item := range roleAssignments.RoleAssignments {
-					roleDefinitionId := path.Base(item.RoleAssignment.Properties.RoleDefinitionId)
-
-					if roleDefinitionId == constants.OwnerRoleID {
-						keyVaultOwner := models.KeyVaultOwner{
-							Owner:      item.RoleAssignment,
-							KeyVaultId: item.KeyVaultId,
-						}
-						log.V(2).Info("found key vault owner", "keyVaultOwner", keyVaultOwner)
-						count++
-						keyVaultOwners.Owners = append(keyVaultOwners.Owners, keyVaultOwner)
-					}
-				}
-				out <- AzureWrapper{
-					Kind: enums.KindAZVMOwner,
-					Data: keyVaultOwners,
-				}
-				log.V(1).Info("finished listing key vault owners", "keyVaultId", roleAssignments.KeyVaultId, "count", count)
+		kvContributors := internal.Map(filteredAssignments, func(ra models.KeyVaultRoleAssignment) models.KeyVaultOwner {
+			return models.KeyVaultOwner{
+				Owner:      ra.RoleAssignment,
+				KeyVaultId: ra.KeyVaultId,
 			}
-		}
-		log.Info("finished listing all key vault owners")
-	}()
+		})
 
-	return out
+		return NewAzureWrapper(enums.KindAZKeyVaultOwner, models.KeyVaultOwners{
+			KeyVaultId: ra.Data.KeyVaultId,
+			Owners:     kvContributors,
+		})
+	})
 }
diff --git a/cmd/list-key-vault-owners_test.go b/cmd/list-key-vault-owners_test.go
index 97a6d5f..2b5c231 100644
--- a/cmd/list-key-vault-owners_test.go
+++ b/cmd/list-key-vault-owners_test.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/bloodhoundad/azurehound/client/mocks"
 	"github.com/bloodhoundad/azurehound/constants"
+	"github.com/bloodhoundad/azurehound/enums"
 	"github.com/bloodhoundad/azurehound/models"
 	"github.com/bloodhoundad/azurehound/models/azure"
 	"github.com/golang/mock/gomock"
@@ -39,16 +40,17 @@ func TestListKeyVaultOwners(t *testing.T) {
 
 	mockClient := mocks.NewMockAzureClient(ctrl)
 
-	mockRoleAssignmentsChannel := make(chan interface{})
+	mockRoleAssignmentsChannel := make(chan azureWrapper[models.KeyVaultRoleAssignments])
 	mockTenant := azure.Tenant{}
 	mockClient.EXPECT().TenantInfo().Return(mockTenant).AnyTimes()
-	channel := listKeyVaultOwners(ctx, mockClient, mockRoleAssignmentsChannel)
+	channel := listKeyVaultOwners(ctx, mockRoleAssignmentsChannel)
 
 	go func() {
 		defer close(mockRoleAssignmentsChannel)
 
-		mockRoleAssignmentsChannel <- AzureWrapper{
-			Data: models.KeyVaultRoleAssignments{
+		mockRoleAssignmentsChannel <- NewAzureWrapper(
+			enums.KindAZKeyVaultRoleAssignment,
+			models.KeyVaultRoleAssignments{
 				KeyVaultId: "foo",
 				RoleAssignments: []models.KeyVaultRoleAssignment{
 					{
@@ -61,15 +63,11 @@ func TestListKeyVaultOwners(t *testing.T) {
 					},
 				},
 			},
-		}
+		)
 	}()
 
-	if result, ok := <-channel; !ok {
+	if _, ok := <-channel; !ok {
 		t.Fatalf("failed to receive from channel")
-	} else if wrapper, ok := result.(AzureWrapper); !ok {
-		t.Errorf("failed type assertion: got %T, want %T", result, AzureWrapper{})
-	} else if _, ok := wrapper.Data.(models.KeyVaultOwners); !ok {
-		t.Errorf("failed type assertion: got %T, want %T", wrapper.Data, models.KeyVaultOwners{})
 	}
 
 	if _, ok := <-channel; ok {
diff --git a/cmd/list-key-vault-role-assignments.go b/cmd/list-key-vault-role-assignments.go
index 208dcc3..ef8f7da 100644
--- a/cmd/list-key-vault-role-assignments.go
+++ b/cmd/list-key-vault-role-assignments.go
@@ -63,9 +63,9 @@ func listKeyVaultRoleAssignmentsCmdImpl(cmd *cobra.Command, args []string) {
 	}
 }
 
-func listKeyVaultRoleAssignments(ctx context.Context, client client.AzureClient, keyVaults <-chan interface{}) <-chan interface{} {
+func listKeyVaultRoleAssignments(ctx context.Context, client client.AzureClient, keyVaults <-chan interface{}) <-chan azureWrapper[models.KeyVaultRoleAssignments] {
 	var (
-		out     = make(chan interface{})
+		out     = make(chan azureWrapper[models.KeyVaultRoleAssignments])
 		ids     = make(chan string)
 		streams = pipeline.Demux(ctx.Done(), ids, 25)
 		wg      sync.WaitGroup
@@ -92,11 +92,11 @@ func listKeyVaultRoleAssignments(ctx context.Context, client client.AzureClient,
 			for id := range stream {
 				var (
 					keyVaultRoleAssignments = models.KeyVaultRoleAssignments{
-						KeyVaultId: id.(string),
+						KeyVaultId: id,
 					}
 					count = 0
 				)
-				for item := range client.ListRoleAssignmentsForResource(ctx, id.(string), "") {
+				for item := range client.ListRoleAssignmentsForResource(ctx, id, "") {
 					if item.Error != nil {
 						log.Error(item.Error, "unable to continue processing role assignments for this key vault", "keyVaultId", id)
 					} else {
@@ -109,10 +109,7 @@ func listKeyVaultRoleAssignments(ctx context.Context, client client.AzureClient,
 						keyVaultRoleAssignments.RoleAssignments = append(keyVaultRoleAssignments.RoleAssignments, keyVaultRoleAssignment)
 					}
 				}
-				out <- AzureWrapper{
-					Kind: enums.KindAZVMRoleAssignment,
-					Data: keyVaultRoleAssignments,
-				}
+				out <- NewAzureWrapper(enums.KindAZKeyVaultRoleAssignment, keyVaultRoleAssignments)
 				log.V(1).Info("finished listing key vault role assignments", "keyVaultId", id, "count", count)
 			}
 		}()
diff --git a/cmd/list-key-vault-role-assignments_test.go b/cmd/list-key-vault-role-assignments_test.go
index 49857ce..aa7aade 100644
--- a/cmd/list-key-vault-role-assignments_test.go
+++ b/cmd/list-key-vault-role-assignments_test.go
@@ -93,21 +93,13 @@ func TestListKeyVaultRoleAssignments(t *testing.T) {
 
 	if result, ok := <-channel; !ok {
 		t.Fatalf("failed to receive from channel")
-	} else if wrapper, ok := result.(AzureWrapper); !ok {
-		t.Errorf("failed type assertion: got %T, want %T", result, AzureWrapper{})
-	} else if data, ok := wrapper.Data.(models.KeyVaultRoleAssignments); !ok {
-		t.Errorf("failed type assertion: got %T, want %T", wrapper.Data, models.KeyVaultRoleAssignments{})
-	} else if len(data.RoleAssignments) != 2 {
-		t.Errorf("got %v, want %v", len(data.RoleAssignments), 2)
+	} else if len(result.Data.RoleAssignments) != 2 {
+		t.Errorf("got %v, want %v", len(result.Data.RoleAssignments), 2)
 	}
 
 	if result, ok := <-channel; !ok {
 		t.Fatalf("failed to receive from channel")
-	} else if wrapper, ok := result.(AzureWrapper); !ok {
-		t.Errorf("failed type assertion: got %T, want %T", result, AzureWrapper{})
-	} else if data, ok := wrapper.Data.(models.KeyVaultRoleAssignments); !ok {
-		t.Errorf("failed type assertion: got %T, want %T", wrapper.Data, models.KeyVaultRoleAssignments{})
-	} else if len(data.RoleAssignments) != 1 {
-		t.Errorf("got %v, want %v", len(data.RoleAssignments), 2)
+	} else if len(result.Data.RoleAssignments) != 1 {
+		t.Errorf("got %v, want %v", len(result.Data.RoleAssignments), 1)
 	}
 }
diff --git a/cmd/list-key-vault-user-access-admins.go b/cmd/list-key-vault-user-access-admins.go
index 4f8a80d..2162c87 100644
--- a/cmd/list-key-vault-user-access-admins.go
+++ b/cmd/list-key-vault-user-access-admins.go
@@ -19,15 +19,13 @@ package cmd
 
 import (
 	"context"
-	"fmt"
 	"os"
 	"os/signal"
-	"path"
 	"time"
 
-	"github.com/bloodhoundad/azurehound/client"
 	"github.com/bloodhoundad/azurehound/constants"
 	"github.com/bloodhoundad/azurehound/enums"
+	"github.com/bloodhoundad/azurehound/internal"
 	"github.com/bloodhoundad/azurehound/models"
 	"github.com/bloodhoundad/azurehound/pipeline"
 	"github.com/spf13/cobra"
@@ -59,52 +57,30 @@ func listKeyVaultUserAccessAdminsCmdImpl(cmd *cobra.Command, args []string) {
 		subscriptions := listSubscriptions(ctx, azClient)
 		keyVaults := listKeyVaults(ctx, azClient, subscriptions)
 		kvRoleAssignments := listKeyVaultRoleAssignments(ctx, azClient, keyVaults)
-		stream := listKeyVaultUserAccessAdmins(ctx, azClient, kvRoleAssignments)
+		stream := listKeyVaultUserAccessAdmins(ctx, kvRoleAssignments)
 		outputStream(ctx, stream)
 		duration := time.Since(start)
 		log.Info("collection completed", "duration", duration.String())
 	}
 }
 
-func listKeyVaultUserAccessAdmins(ctx context.Context, client client.AzureClient, vmRoleAssignments <-chan interface{}) <-chan interface{} {
-	out := make(chan interface{})
+func listKeyVaultUserAccessAdmins(
+	ctx context.Context,
+	kvRoleAssignments <-chan azureWrapper[models.KeyVaultRoleAssignments],
+) <-chan any {
+	return pipeline.Map(ctx.Done(), kvRoleAssignments, func(ra azureWrapper[models.KeyVaultRoleAssignments]) any {
+		filteredAssignments := internal.Filter(ra.Data.RoleAssignments, kvRoleAssignmentFilter(constants.UserAccessAdminRoleID))
 
-	go func() {
-		defer close(out)
-
-		for result := range pipeline.OrDone(ctx.Done(), vmRoleAssignments) {
-			if roleAssignments, ok := result.(AzureWrapper).Data.(models.KeyVaultRoleAssignments); !ok {
-				log.Error(fmt.Errorf("failed type assertion"), "unable to continue enumerating key vault userAccessAdmins", "result", result)
-				return
-			} else {
-				var (
-					keyVaultUserAccessAdmins = models.KeyVaultUserAccessAdmins{
-						KeyVaultId: roleAssignments.KeyVaultId,
-					}
-					count = 0
-				)
-				for _, item := range roleAssignments.RoleAssignments {
-					roleDefinitionId := path.Base(item.RoleAssignment.Properties.RoleDefinitionId)
-
-					if roleDefinitionId == constants.UserAccessAdminRoleID {
-						keyVaultUserAccessAdmin := models.KeyVaultUserAccessAdmin{
-							UserAccessAdmin: item.RoleAssignment,
-							KeyVaultId:      item.KeyVaultId,
-						}
-						log.V(2).Info("found key vault userAccessAdmin", "keyVaultUserAccessAdmin", keyVaultUserAccessAdmin)
-						count++
-						keyVaultUserAccessAdmins.UserAccessAdmins = append(keyVaultUserAccessAdmins.UserAccessAdmins, keyVaultUserAccessAdmin)
-					}
-				}
-				out <- AzureWrapper{
-					Kind: enums.KindAZVMUserAccessAdmin,
-					Data: keyVaultUserAccessAdmins,
-				}
-				log.V(1).Info("finished listing key vault userAccessAdmins", "keyVaultId", roleAssignments.KeyVaultId, "count", count)
+		kvContributors := internal.Map(filteredAssignments, func(ra models.KeyVaultRoleAssignment) models.KeyVaultUserAccessAdmin {
+			return models.KeyVaultUserAccessAdmin{
+				UserAccessAdmin: ra.RoleAssignment,
+				KeyVaultId:      ra.KeyVaultId,
 			}
-		}
-		log.Info("finished listing all key vault userAccessAdmins")
-	}()
+		})
 
-	return out
+		return NewAzureWrapper(enums.KindAZKeyVaultUserAccessAdmin, models.KeyVaultUserAccessAdmins{
+			KeyVaultId:       ra.Data.KeyVaultId,
+			UserAccessAdmins: kvContributors,
+		})
+	})
 }
diff --git a/cmd/list-key-vault-user-access-admins_test.go b/cmd/list-key-vault-user-access-admins_test.go
index 670a151..d95449c 100644
--- a/cmd/list-key-vault-user-access-admins_test.go
+++ b/cmd/list-key-vault-user-access-admins_test.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/bloodhoundad/azurehound/client/mocks"
 	"github.com/bloodhoundad/azurehound/constants"
+	"github.com/bloodhoundad/azurehound/enums"
 	"github.com/bloodhoundad/azurehound/models"
 	"github.com/bloodhoundad/azurehound/models/azure"
 	"github.com/golang/mock/gomock"
@@ -39,16 +40,17 @@ func TestListKeyVaultUserAccessAdmins(t *testing.T) {
 
 	mockClient := mocks.NewMockAzureClient(ctrl)
 
-	mockRoleAssignmentsChannel := make(chan interface{})
+	mockRoleAssignmentsChannel := make(chan azureWrapper[models.KeyVaultRoleAssignments])
 	mockTenant := azure.Tenant{}
 	mockClient.EXPECT().TenantInfo().Return(mockTenant).AnyTimes()
-	channel := listKeyVaultUserAccessAdmins(ctx, mockClient, mockRoleAssignmentsChannel)
+	channel := listKeyVaultUserAccessAdmins(ctx, mockRoleAssignmentsChannel)
 
 	go func() {
 		defer close(mockRoleAssignmentsChannel)
 
-		mockRoleAssignmentsChannel <- AzureWrapper{
-			Data: models.KeyVaultRoleAssignments{
+		mockRoleAssignmentsChannel <- NewAzureWrapper(
+			enums.KindAZKeyVaultRoleAssignment,
+			models.KeyVaultRoleAssignments{
 				KeyVaultId: "foo",
 				RoleAssignments: []models.KeyVaultRoleAssignment{
 					{
@@ -61,15 +63,11 @@ func TestListKeyVaultUserAccessAdmins(t *testing.T) {
 					},
 				},
 			},
-		}
+		)
 	}()
 
-	if result, ok := <-channel; !ok {
+	if _, ok := <-channel; !ok {
 		t.Fatalf("failed to receive from channel")
-	} else if wrapper, ok := result.(AzureWrapper); !ok {
-		t.Errorf("failed type assertion: got %T, want %T", result, AzureWrapper{})
-	} else if _, ok := wrapper.Data.(models.KeyVaultUserAccessAdmins); !ok {
-		t.Errorf("failed type assertion: got %T, want %T", wrapper.Data, models.KeyVaultUserAccessAdmins{})
 	}
 
 	if _, ok := <-channel; ok {
diff --git a/cmd/list-key-vaults.go b/cmd/list-key-vaults.go
index cee7720..a8d582c 100644
--- a/cmd/list-key-vaults.go
+++ b/cmd/list-key-vaults.go
@@ -90,7 +90,7 @@ func listKeyVaults(ctx context.Context, client client.AzureClient, subscriptions
 			defer wg.Done()
 			for id := range stream {
 				count := 0
-				for item := range client.ListAzureKeyVaults(ctx, id.(string), 999) {
+				for item := range client.ListAzureKeyVaults(ctx, id, 999) {
 					if item.Error != nil {
 						log.Error(item.Error, "unable to continue processing key vaults for this subscription", "subscriptionId", id)
 					} else {
diff --git a/cmd/list-management-group-descendants.go b/cmd/list-management-group-descendants.go
index 31350f2..e04208e 100644
--- a/cmd/list-management-group-descendants.go
+++ b/cmd/list-management-group-descendants.go
@@ -90,7 +90,7 @@ func listManagementGroupDescendants(ctx context.Context, client client.AzureClie
 			defer wg.Done()
 			for id := range stream {
 				count := 0
-				for item := range client.ListAzureManagementGroupDescendants(ctx, id.(string)) {
+				for item := range client.ListAzureManagementGroupDescendants(ctx, id) {
 					if item.Error != nil {
 						log.Error(item.Error, "unable to continue processing descendants for this management group", "managementGroupId", id)
 					} else {
diff --git a/cmd/list-management-group-owners.go b/cmd/list-management-group-owners.go
index 73f6c19..12f7fd7 100644
--- a/cmd/list-management-group-owners.go
+++ b/cmd/list-management-group-owners.go
@@ -93,11 +93,11 @@ func listManagementGroupOwners(ctx context.Context, client client.AzureClient, m
 			for id := range stream {
 				var (
 					managementGroupOwners = models.ManagementGroupOwners{
-						ManagementGroupId: id.(string),
+						ManagementGroupId: id,
 					}
 					count = 0
 				)
-				for item := range client.ListRoleAssignmentsForResource(ctx, id.(string), "") {
+				for item := range client.ListRoleAssignmentsForResource(ctx, id, "") {
 					if item.Error != nil {
 						log.Error(item.Error, "unable to continue processing owners for this management group", "managementGroupId", id)
 					} else {
diff --git a/cmd/list-management-group-user-access-admins.go b/cmd/list-management-group-user-access-admins.go
index e990571..b207245 100644
--- a/cmd/list-management-group-user-access-admins.go
+++ b/cmd/list-management-group-user-access-admins.go
@@ -94,11 +94,11 @@ func listManagementGroupUserAccessAdmins(ctx context.Context, client client.Azur
 			for id := range stream {
 				var (
 					mgmtGroupUserAccessAdmins = models.ManagementGroupUserAccessAdmins{
-						ManagementGroupId: id.(string),
+						ManagementGroupId: id,
 					}
 					count = 0
 				)
-				for item := range client.ListRoleAssignmentsForResource(ctx, id.(string), "") {
+				for item := range client.ListRoleAssignmentsForResource(ctx, id, "") {
 					if item.Error != nil {
 						log.Error(item.Error, "unable to continue processing user access admins for this management group", "managementGroupId", id)
 					} else {
diff --git a/cmd/list-resource-group-owners.go b/cmd/list-resource-group-owners.go
index 1a86a8d..fd6923f 100644
--- a/cmd/list-resource-group-owners.go
+++ b/cmd/list-resource-group-owners.go
@@ -94,11 +94,11 @@ func listResourceGroupOwners(ctx context.Context, client client.AzureClient, res
 			for id := range stream {
 				var (
 					resourceGroupOwners = models.ResourceGroupOwners{
-						ResourceGroupId: id.(string),
+						ResourceGroupId: id,
 					}
 					count = 0
 				)
-				for item := range client.ListRoleAssignmentsForResource(ctx, id.(string), "") {
+				for item := range client.ListRoleAssignmentsForResource(ctx, id, "") {
 					if item.Error != nil {
 						log.Error(item.Error, "unable to continue processing owners for this resource group", "resourceGroupId", id)
 					} else {
diff --git a/cmd/list-resource-group-user-access-admins.go b/cmd/list-resource-group-user-access-admins.go
index a89f6d9..7870d64 100644
--- a/cmd/list-resource-group-user-access-admins.go
+++ b/cmd/list-resource-group-user-access-admins.go
@@ -94,11 +94,11 @@ func listResourceGroupUserAccessAdmins(ctx context.Context, client client.AzureC
 			for id := range stream {
 				var (
 					resourceGroupUserAccessAdmins = models.ResourceGroupUserAccessAdmins{
-						ResourceGroupId: id.(string),
+						ResourceGroupId: id,
 					}
 					count = 0
 				)
-				for item := range client.ListRoleAssignmentsForResource(ctx, id.(string), "") {
+				for item := range client.ListRoleAssignmentsForResource(ctx, id, "") {
 					if item.Error != nil {
 						log.Error(item.Error, "unable to continue processing user access admins for this resource group", "resourceGroupId", id)
 					} else {
diff --git a/cmd/list-resource-groups.go b/cmd/list-resource-groups.go
index ef225eb..cbba504 100644
--- a/cmd/list-resource-groups.go
+++ b/cmd/list-resource-groups.go
@@ -90,7 +90,7 @@ func listResourceGroups(ctx context.Context, client client.AzureClient, subscrip
 			defer wg.Done()
 			for id := range stream {
 				count := 0
-				for item := range client.ListAzureResourceGroups(ctx, id.(string), "") {
+				for item := range client.ListAzureResourceGroups(ctx, id, "") {
 					if item.Error != nil {
 						log.Error(item.Error, "unable to continue processing resource groups for this subscription", "subscriptionId", id)
 					} else {
diff --git a/cmd/list-role-assignments.go b/cmd/list-role-assignments.go
index c3a53c7..2a4ab98 100644
--- a/cmd/list-role-assignments.go
+++ b/cmd/list-role-assignments.go
@@ -92,11 +92,11 @@ func listRoleAssignments(ctx context.Context, client client.AzureClient, roles <
 			for id := range stream {
 				var (
 					roleAssignments = models.RoleAssignments{
-						RoleDefinitionId: id.(string),
+						RoleDefinitionId: id,
 						TenantId:         client.TenantInfo().TenantId,
 					}
 					count  = 0
-					filter = fmt.Sprintf("roleDefinitionId eq '%s'", id.(string))
+					filter = fmt.Sprintf("roleDefinitionId eq '%s'", id)
 				)
 				for item := range client.ListAzureADRoleAssignments(ctx, filter, "", "", "", nil) {
 					if item.Error != nil {
diff --git a/cmd/list-root.go b/cmd/list-root.go
index bf36080..274d4f2 100644
--- a/cmd/list-root.go
+++ b/cmd/list-root.go
@@ -27,6 +27,7 @@ import (
 	"github.com/bloodhoundad/azurehound/client"
 	"github.com/bloodhoundad/azurehound/config"
 	"github.com/bloodhoundad/azurehound/enums"
+	"github.com/bloodhoundad/azurehound/models"
 	"github.com/bloodhoundad/azurehound/pipeline"
 	"github.com/spf13/cobra"
 )
@@ -82,9 +83,11 @@ func listAll(ctx context.Context, client client.AzureClient) <-chan interface{}
 		keyVaults  = make(chan interface{})
 		keyVaults2 = make(chan interface{})
 		keyVaults3 = make(chan interface{})
-		keyVaults4 = make(chan interface{})
-		keyVaults5 = make(chan interface{})
-		keyVaults6 = make(chan interface{})
+
+		keyVaultRoleAssignments1 = make(chan azureWrapper[models.KeyVaultRoleAssignments])
+		keyVaultRoleAssignments2 = make(chan azureWrapper[models.KeyVaultRoleAssignments])
+		keyVaultRoleAssignments3 = make(chan azureWrapper[models.KeyVaultRoleAssignments])
+		keyVaultRoleAssignments4 = make(chan azureWrapper[models.KeyVaultRoleAssignments])
 
 		mgmtGroups  = make(chan interface{})
 		mgmtGroups2 = make(chan interface{})
@@ -140,12 +143,14 @@ func listAll(ctx context.Context, client client.AzureClient) <-chan interface{}
 	subscriptionUserAccessAdmins := listSubscriptionUserAccessAdmins(ctx, client, subscriptions6)
 
 	// Enumerate KeyVaults, KeyVaultOwners, KeyVaultAccessPolicies and KeyVaultUserAccessAdmins
-	pipeline.Tee(ctx.Done(), listKeyVaults(ctx, client, subscriptions2), keyVaults, keyVaults2, keyVaults3, keyVaults4, keyVaults5, keyVaults6)
-	keyVaultOwners := listKeyVaultOwners(ctx, client, keyVaults2)
+	pipeline.Tee(ctx.Done(), listKeyVaults(ctx, client, subscriptions2), keyVaults, keyVaults2, keyVaults3)
+	pipeline.Tee(ctx.Done(), listKeyVaultRoleAssignments(ctx, client, keyVaults2), keyVaultRoleAssignments1, keyVaultRoleAssignments2, keyVaultRoleAssignments3, keyVaultRoleAssignments4)
 	keyVaultAccessPolicies := listKeyVaultAccessPolicies(ctx, client, keyVaults3, []enums.KeyVaultAccessType{enums.GetCerts, enums.GetKeys, enums.GetCerts})
-	keyVaultUserAccessAdmins := listKeyVaultUserAccessAdmins(ctx, client, keyVaults4)
-	keyVaultContributors := listKeyVaultContributors(ctx, client, keyVaults5)
-	keyVaultKVContributors := listKeyVaultKVContributors(ctx, client, keyVaults6)
+
+	keyVaultOwners := listKeyVaultOwners(ctx, keyVaultRoleAssignments1)
+	keyVaultUserAccessAdmins := listKeyVaultUserAccessAdmins(ctx, keyVaultRoleAssignments2)
+	keyVaultContributors := listKeyVaultContributors(ctx, keyVaultRoleAssignments3)
+	keyVaultKVContributors := listKeyVaultKVContributors(ctx, keyVaultRoleAssignments4)
 
 	// Enumerate ManagementGroups, ManagementGroupOwners and ManagementGroupDescendants
 	pipeline.Tee(ctx.Done(), listManagementGroups(ctx, client), mgmtGroups, mgmtGroups2, mgmtGroups3, mgmtGroups4)
diff --git a/cmd/list-service-principal-owners.go b/cmd/list-service-principal-owners.go
index bd0bf12..e856bdf 100644
--- a/cmd/list-service-principal-owners.go
+++ b/cmd/list-service-principal-owners.go
@@ -91,11 +91,11 @@ func listServicePrincipalOwners(ctx context.Context, client client.AzureClient,
 			for id := range stream {
 				var (
 					servicePrincipalOwners = models.ServicePrincipalOwners{
-						ServicePrincipalId: id.(string),
+						ServicePrincipalId: id,
 					}
 					count = 0
 				)
-				for item := range client.ListAzureADServicePrincipalOwners(ctx, id.(string), "", "", "", nil) {
+				for item := range client.ListAzureADServicePrincipalOwners(ctx, id, "", "", "", nil) {
 					if item.Error != nil {
 						log.Error(item.Error, "unable to continue processing owners for this service principal", "servicePrincipalId", id)
 					} else {
diff --git a/cmd/list-subscription-owners.go b/cmd/list-subscription-owners.go
index 5789bf4..0dd2baf 100644
--- a/cmd/list-subscription-owners.go
+++ b/cmd/list-subscription-owners.go
@@ -94,11 +94,11 @@ func listSubscriptionOwners(ctx context.Context, client client.AzureClient, subs
 			for id := range stream {
 				var (
 					subscriptionOwners = models.SubscriptionOwners{
-						SubscriptionId: id.(string),
+						SubscriptionId: id,
 					}
 					count = 0
 				)
-				for item := range client.ListRoleAssignmentsForResource(ctx, id.(string), "") {
+				for item := range client.ListRoleAssignmentsForResource(ctx, id, "") {
 					if item.Error != nil {
 						log.Error(item.Error, "unable to continue processing owners for this subscription", "subscriptionId", id)
 					} else {
diff --git a/cmd/list-subscription-user-access-admins.go b/cmd/list-subscription-user-access-admins.go
index 70734a6..5077399 100644
--- a/cmd/list-subscription-user-access-admins.go
+++ b/cmd/list-subscription-user-access-admins.go
@@ -94,11 +94,11 @@ func listSubscriptionUserAccessAdmins(ctx context.Context, client client.AzureCl
 			for id := range stream {
 				var (
 					subscriptionUserAccessAdmins = models.SubscriptionUserAccessAdmins{
-						SubscriptionId: id.(string),
+						SubscriptionId: id,
 					}
 					count = 0
 				)
-				for item := range client.ListRoleAssignmentsForResource(ctx, id.(string), "") {
+				for item := range client.ListRoleAssignmentsForResource(ctx, id, "") {
 					if item.Error != nil {
 						log.Error(item.Error, "unable to continue processing user access admins for this subscription", "subscriptionId", id)
 					} else {
diff --git a/cmd/list-virtual-machine-role-assignments.go b/cmd/list-virtual-machine-role-assignments.go
index a861865..5ae32f8 100644
--- a/cmd/list-virtual-machine-role-assignments.go
+++ b/cmd/list-virtual-machine-role-assignments.go
@@ -92,11 +92,11 @@ func listVirtualMachineRoleAssignments(ctx context.Context, client client.AzureC
 			for id := range stream {
 				var (
 					virtualMachineRoleAssignments = models.VirtualMachineRoleAssignments{
-						VirtualMachineId: id.(string),
+						VirtualMachineId: id,
 					}
 					count = 0
 				)
-				for item := range client.ListRoleAssignmentsForResource(ctx, id.(string), "") {
+				for item := range client.ListRoleAssignmentsForResource(ctx, id, "") {
 					if item.Error != nil {
 						log.Error(item.Error, "unable to continue processing role assignments for this virtual machine", "virtualMachineId", id)
 					} else {
diff --git a/cmd/list-virtual-machines.go b/cmd/list-virtual-machines.go
index 31a147c..9483cfd 100644
--- a/cmd/list-virtual-machines.go
+++ b/cmd/list-virtual-machines.go
@@ -89,7 +89,7 @@ func listVirtualMachines(ctx context.Context, client client.AzureClient, subscri
 			defer wg.Done()
 			for id := range stream {
 				count := 0
-				for item := range client.ListAzureVirtualMachines(ctx, id.(string), false) {
+				for item := range client.ListAzureVirtualMachines(ctx, id, false) {
 					if item.Error != nil {
 						log.Error(item.Error, "unable to continue processing virtual machines for this subscription", "subscriptionId", id)
 					} else {
diff --git a/cmd/utils.go b/cmd/utils.go
index 35ca337..56d044d 100644
--- a/cmd/utils.go
+++ b/cmd/utils.go
@@ -32,6 +32,7 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"path"
 	"path/filepath"
 	"time"
 
@@ -41,6 +42,7 @@ import (
 	"github.com/bloodhoundad/azurehound/config"
 	"github.com/bloodhoundad/azurehound/enums"
 	"github.com/bloodhoundad/azurehound/logger"
+	"github.com/bloodhoundad/azurehound/models"
 	"github.com/bloodhoundad/azurehound/pipeline"
 	"github.com/bloodhoundad/azurehound/sinks"
 	"github.com/spf13/cobra"
@@ -371,12 +373,25 @@ func setupLogger() {
 	}
 }
 
+// deprecated: use azureWrapper instead
 type AzureWrapper struct {
 	Kind enums.Kind  `json:"kind"`
 	Data interface{} `json:"data"`
 }
 
-func outputStream(ctx context.Context, stream <-chan interface{}) {
+type azureWrapper[T any] struct {
+	Kind enums.Kind `json:"kind"`
+	Data T          `json:"data"`
+}
+
+func NewAzureWrapper[T any](kind enums.Kind, data T) azureWrapper[T] {
+	return azureWrapper[T]{
+		Kind: kind,
+		Data: data,
+	}
+}
+
+func outputStream[T any](ctx context.Context, stream <-chan T) {
 	formatted := pipeline.FormatJson(ctx.Done(), stream)
 	if path := config.OutputFile.Value().(string); path != "" {
 		if err := sinks.WriteToFile(ctx, path, formatted); err != nil {
@@ -386,3 +401,9 @@ func outputStream(ctx context.Context, stream <-chan interface{}) {
 		sinks.WriteToConsole(ctx, formatted)
 	}
 }
+
+func kvRoleAssignmentFilter(roleId string) func(models.KeyVaultRoleAssignment) bool {
+	return func(ra models.KeyVaultRoleAssignment) bool {
+		return path.Base(ra.RoleAssignment.Properties.RoleDefinitionId) == roleId
+	}
+}
diff --git a/enums/kind.go b/enums/kind.go
index b2380ac..ad05241 100644
--- a/enums/kind.go
+++ b/enums/kind.go
@@ -33,6 +33,7 @@ const (
 	KindAZKeyVaultContributor            Kind = "AZKeyVaultContributor"
 	KindAZKeyVaultKVContributor          Kind = "AZKeyVaultKVContributor"
 	KindAZKeyVaultOwner                  Kind = "AZKeyVaultOwner"
+	KindAZKeyVaultRoleAssignment         Kind = "AZKeyVaultRoleAssignment"
 	KindAZKeyVaultUserAccessAdmin        Kind = "AZKeyVaultUserAccessAdmin"
 	KindAZManagementGroup                Kind = "AZManagementGroup"
 	KindAZManagementGroupOwner           Kind = "AZManagementGroupOwner"
diff --git a/internal/utils.go b/internal/utils.go
new file mode 100644
index 0000000..f7ce20e
--- /dev/null
+++ b/internal/utils.go
@@ -0,0 +1,19 @@
+package internal
+
+func Map[T, U any](collection []T, fn func(T) U) []U {
+	var out []U
+	for i := range collection {
+		out = append(out, fn(collection[i]))
+	}
+	return out
+}
+
+func Filter[T any](collection []T, fn func(T) bool) []T {
+	var out []T
+	for i := range collection {
+		if fn(collection[i]) {
+			out = append(out, collection[i])
+		}
+	}
+	return out
+}
diff --git a/pipeline/pipeline.go b/pipeline/pipeline.go
new file mode 100644
index 0000000..f3110b0
--- /dev/null
+++ b/pipeline/pipeline.go
@@ -0,0 +1,229 @@
+// Copyright (C) 2022 Specter Ops, Inc.
+//
+// This file is part of AzureHound.
+//
+// AzureHound is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// AzureHound is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+package pipeline
+
+import (
+	"encoding/json"
+	"reflect"
+	"sync"
+	"time"
+
+	"github.com/bloodhoundad/azurehound/internal"
+)
+
+type Result[T any] struct {
+	Error error
+	Ok    T
+}
+
+// OrDone provides an explicit cancellation mechanism to ensure the encapsulated and downstream goroutines are cleaned
+// up. This frees the caller from depending on the input channel to close in order to free the goroutine, thus
+// preventing possible leaks.
+func OrDone[D, T any](done <-chan D, in <-chan T) <-chan T {
+	out := make(chan T)
+
+	go func() {
+		defer close(out)
+		for {
+			select {
+			case <-done:
+				return
+			case val, ok := <-in:
+				if !ok {
+					return
+				} else {
+					select {
+					case out <- val:
+					case <-done:
+					}
+				}
+			}
+		}
+	}()
+	return out
+}
+
+// Mux joins multiple channels and returns a channel as single stream of data.
+func Mux[D any](done <-chan D, channels ...<-chan any) <-chan any {
+	var wg sync.WaitGroup
+	out := make(chan interface{})
+
+	muxer := func(channel <-chan any) {
+		defer wg.Done()
+		for item := range OrDone(done, channel) {
+			out <- item
+		}
+	}
+
+	wg.Add(len(channels))
+	for _, channel := range channels {
+		go muxer(channel)
+	}
+
+	go func() {
+		wg.Wait()
+		close(out)
+	}()
+
+	return out
+}
+
+// Demux distributes the stream of data from a single channel across multiple channels to parallelize CPU use and I/O
+func Demux[D, T any](done <-chan D, in <-chan T, size int) []<-chan T {
+	outputs := make([]chan T, size)
+
+	for i := range outputs {
+		outputs[i] = make(chan T)
+	}
+
+	closeOutputs := func() {
+		for i := range outputs {
+			close(outputs[i])
+		}
+	}
+
+	cases := internal.Map(outputs, func(out chan T) reflect.SelectCase {
+		return reflect.SelectCase{
+			Dir:  reflect.SelectSend,
+			Chan: reflect.ValueOf(out),
+		}
+	})
+
+	go func() {
+		defer closeOutputs()
+		for item := range OrDone(done, in) {
+			// send item to exactly one channel
+			for i := range cases {
+				cases[i].Send = reflect.ValueOf(item)
+			}
+			reflect.Select(cases)
+		}
+	}()
+
+	return internal.Map(outputs, func(out chan T) <-chan T { return out })
+}
+
+func Map[D, T, U any](done <-chan D, in <-chan T, fn func(T) U) <-chan U {
+	out := make(chan U)
+	go func() {
+		defer close(out)
+		for item := range OrDone(done, in) {
+			out <- fn(item)
+		}
+	}()
+	return out
+}
+
+func Filter[D, T any](done <-chan D, in <-chan T, fn func(T) bool) <-chan T {
+	out := make(chan T)
+	go func() {
+		defer close(out)
+		for item := range OrDone(done, in) {
+			if fn(item) {
+				out <- item
+			}
+		}
+	}()
+	return out
+}
+
+// Tee copies the stream of data from a single channel to zero or more channels
+func Tee[D, T any](done <-chan D, in <-chan T, outputs ...chan<- T) {
+	go func() {
+		// Need to close outputs when goroutine exits to ensure we avoid deadlock
+		defer func() {
+			for i := range outputs {
+				close(outputs[i])
+			}
+		}()
+
+		for item := range OrDone(done, in) {
+			for _, out := range outputs {
+				select {
+				case <-done:
+				case out <- item:
+				}
+			}
+		}
+	}()
+}
+
+func Batch[D, T any](done <-chan D, in <-chan T, maxItems int, maxTimeout time.Duration) <-chan []T {
+	out := make(chan []T)
+
+	go func() {
+		defer close(out)
+
+		timeout := time.After(maxTimeout)
+		var batch []T
+		for {
+			select {
+			case <-done:
+				if len(batch) > 0 {
+					out <- batch
+					batch = nil
+				}
+				return
+			case item, ok := <-in:
+				if !ok {
+					if len(batch) > 0 {
+						out <- batch
+						batch = nil
+					}
+					return
+				} else {
+					// Add to batch
+					batch = append(batch, item)
+
+					// Flush if limit is reached
+					if len(batch) >= maxItems {
+						out <- batch
+						batch = nil
+						timeout = time.After(maxTimeout)
+					}
+				}
+			case <-timeout:
+				if len(batch) > 0 {
+					out <- batch
+					batch = nil
+				}
+				timeout = time.After(maxTimeout)
+			}
+		}
+	}()
+
+	return out
+}
+
+func FormatJson[D, T any](done <-chan D, in <-chan T) <-chan string {
+	out := make(chan string)
+
+	go func() {
+		defer close(out)
+
+		for item := range OrDone(done, in) {
+			if bytes, err := json.Marshal(item); err != nil {
+				panic(err)
+			} else {
+				out <- string(bytes)
+			}
+		}
+	}()
+
+	return out
+}
diff --git a/pipeline/utils_test.go b/pipeline/pipeline_test.go
similarity index 100%
rename from pipeline/utils_test.go
rename to pipeline/pipeline_test.go
diff --git a/pipeline/utils.go b/pipeline/utils.go
deleted file mode 100644
index 76021bf..0000000
--- a/pipeline/utils.go
+++ /dev/null
@@ -1,262 +0,0 @@
-// Copyright (C) 2022 Specter Ops, Inc.
-//
-// This file is part of AzureHound.
-//
-// AzureHound is free software: you can redistribute it and/or modify
-// it under the terms of the GNU General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// AzureHound is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-// GNU General Public License for more details.
-//
-// You should have received a copy of the GNU General Public License
-// along with this program.  If not, see <https://www.gnu.org/licenses/>.
-
-package pipeline
-
-import (
-	"encoding/json"
-	"fmt"
-	"reflect"
-	"sync"
-	"time"
-)
-
-type Result struct {
-	Error error
-	Ok    interface{}
-}
-
-// OrDone provides an explicit cancellation mechanism to ensure the encapsulated and downstream goroutines are cleaned
-// up. This frees the caller from depending on the input channel to close in order to free the goroutine, thus
-// preventing possible leaks.
-func OrDone(done, in interface{}) <-chan interface{} {
-	if !isReadable(done) || !isReadable(in) {
-		panic(fmt.Errorf("channels must be readable"))
-	}
-	out := make(chan interface{})
-
-	go func() {
-		defer close(out)
-		doneCase := reflect.SelectCase{
-			Dir:  reflect.SelectRecv,
-			Chan: reflect.ValueOf(done),
-		}
-
-		outerCases := []reflect.SelectCase{
-			doneCase,
-			{
-				Dir:  reflect.SelectRecv,
-				Chan: reflect.ValueOf(in),
-			},
-		}
-
-		innerCases := []reflect.SelectCase{
-			doneCase,
-			{
-				Dir:  reflect.SelectSend,
-				Chan: reflect.ValueOf(out),
-			},
-		}
-		for {
-			if chosen, item, ok := reflect.Select(outerCases); chosen == 0 || !ok {
-				// If received on done then return
-				return
-			} else {
-				if !ok {
-					return
-				} else {
-					innerCases[1].Send = item
-					if chosen, _, _ := reflect.Select(innerCases); chosen == 0 || !ok {
-						return
-					}
-				}
-			}
-		}
-	}()
-	return out
-}
-
-// Mux joins multiple channels and returns a channel as single stream of data.
-func Mux(done interface{}, channels ...interface{}) <-chan interface{} {
-	var wg sync.WaitGroup
-	out := make(chan interface{})
-
-	muxer := func(channel interface{}) {
-		defer wg.Done()
-		for item := range OrDone(done, channel) {
-			out <- item
-		}
-	}
-
-	wg.Add(len(channels))
-	for _, channel := range channels {
-		go muxer(channel)
-	}
-
-	go func() {
-		wg.Wait()
-		close(out)
-	}()
-
-	return out
-}
-
-// Demux distributes the stream of data from a single channel across multiple channels to parallelize CPU use and I/O
-func Demux(done interface{}, in interface{}, size int) []<-chan interface{} {
-	// use reflection to dynamically create select statement
-	outputs := make([]chan interface{}, size)
-	readChans := []<-chan interface{}{}
-	for i := range outputs {
-		out := make(chan interface{})
-		outputs[i] = out
-		readChans = append(readChans, out)
-	}
-
-	closeOutputs := func() {
-		for i := range outputs {
-			close(outputs[i])
-		}
-	}
-
-	cases := make([]reflect.SelectCase, len(outputs))
-	for i := range cases {
-		cases[i].Dir = reflect.SelectSend
-		cases[i].Chan = reflect.ValueOf(outputs[i])
-	}
-	cases = append(cases, reflect.SelectCase{
-		Dir:  reflect.SelectRecv,
-		Chan: reflect.ValueOf(done),
-	})
-
-	go func() {
-		defer closeOutputs()
-		for item := range OrDone(done, in) {
-			// send item to exactly once channel or cancel
-			for i := range cases {
-				if cases[i].Dir == reflect.SelectSend {
-					cases[i].Send = reflect.ValueOf(item)
-				}
-			}
-
-			reflect.Select(cases)
-		}
-	}()
-
-	return readChans
-}
-
-// Tee copies the stream of data from a single channel to zero or more channels
-func Tee(done interface{}, in interface{}, outputs ...chan<- interface{}) {
-	// use reflection to dynamically create select block
-	cases := make([]reflect.SelectCase, len(outputs))
-	for i := range cases {
-		cases[i].Dir = reflect.SelectSend
-	}
-
-	go func() {
-		// Need to close outputs when goroutine exits to ensure we avoid deadlock
-		defer func() {
-			for i := range outputs {
-				close(outputs[i])
-			}
-		}()
-
-		for item := range OrDone(done, in) {
-			// setup all possible select cases
-			for i := range cases {
-				cases[i].Chan = reflect.ValueOf(outputs[i])
-				cases[i].Send = reflect.ValueOf(item)
-			}
-
-			// send item to each channel no more than once or cancel
-			for range cases {
-				chosen, _, _ := reflect.Select(cases)
-				cases[chosen].Chan = reflect.ValueOf(nil)
-			}
-		}
-	}()
-}
-
-func Batch(done interface{}, in interface{}, maxItems int, maxTimeout time.Duration) <-chan []interface{} {
-	if !isReadable(done) || !isReadable(in) {
-		panic(fmt.Errorf("channels must be readable"))
-	}
-	out := make(chan []interface{})
-
-	go func() {
-		defer close(out)
-
-		doneCase := reflect.SelectCase{
-			Dir:  reflect.SelectRecv,
-			Chan: reflect.ValueOf(done),
-		}
-
-		itemCase := reflect.SelectCase{
-			Dir:  reflect.SelectRecv,
-			Chan: reflect.ValueOf(in),
-		}
-
-		timeoutCase := reflect.SelectCase{
-			Dir:  reflect.SelectRecv,
-			Chan: reflect.ValueOf(time.After(maxTimeout)),
-		}
-
-		var batch []interface{}
-		for {
-			if chosen, item, ok := reflect.Select([]reflect.SelectCase{doneCase, itemCase, timeoutCase}); chosen == 0 || !ok {
-				// Flush and return when canceled or closed
-				if len(batch) > 0 {
-					out <- batch
-					batch = nil
-				}
-				return
-			} else if chosen == 1 {
-				// Add to batch
-				batch = append(batch, item.Interface())
-
-				// Flush if limit is reached
-				if len(batch) >= maxItems {
-					out <- batch
-					batch = nil
-					timeoutCase.Chan = reflect.ValueOf(time.After(maxTimeout))
-				}
-			} else {
-				// Timeout triggered, flush and reset
-				if len(batch) > 0 {
-					out <- batch
-					batch = nil
-				}
-				timeoutCase.Chan = reflect.ValueOf(time.After(maxTimeout))
-			}
-		}
-	}()
-
-	return out
-}
-
-func FormatJson(done interface{}, in interface{}) <-chan interface{} {
-	out := make(chan interface{})
-
-	go func() {
-		defer close(out)
-
-		for item := range OrDone(done, in) {
-			if bytes, err := json.Marshal(item); err != nil {
-				panic(err)
-			} else {
-				out <- string(bytes)
-			}
-		}
-	}()
-
-	return out
-}
-
-func isReadable(channel interface{}) bool {
-	channelType := reflect.TypeOf(channel)
-	return channelType.Kind() == reflect.Chan && channelType.ChanDir() != reflect.SendDir
-}
diff --git a/sinks/console.go b/sinks/console.go
index 8c24e1a..db68a09 100644
--- a/sinks/console.go
+++ b/sinks/console.go
@@ -24,7 +24,7 @@ import (
 	"github.com/bloodhoundad/azurehound/pipeline"
 )
 
-func WriteToConsole(ctx context.Context, stream <-chan interface{}) {
+func WriteToConsole[T any](ctx context.Context, stream <-chan T) {
 	for item := range pipeline.OrDone(ctx.Done(), stream) {
 		fmt.Println(item)
 	}
diff --git a/sinks/file.go b/sinks/file.go
index 8dcf4ec..3d8eea6 100644
--- a/sinks/file.go
+++ b/sinks/file.go
@@ -27,7 +27,7 @@ import (
 	"github.com/bloodhoundad/azurehound/pipeline"
 )
 
-func WriteToFile(ctx context.Context, filePath string, stream <-chan interface{}) error {
+func WriteToFile[T any](ctx context.Context, filePath string, stream <-chan T) error {
 
 	if file, err := os.OpenFile(filePath, os.O_CREATE|os.O_WRONLY, 0666); err != nil {
 		return err