x-pack/filebeat/input/entityanalytics/provider/azuread: add registere…

…d owner/user handling (#36092)

CHANGELOG.next.asciidoc

@@ -356,6 +356,7 @@ automatic splitting at root level, if root level element is an array. {pull}3415
 - Add fingerprint mode for the filestream scanner and new file identity based on it {issue}34419[34419] {pull}35734[35734]
 - Add file system metadata to events ingested via filestream {issue}35801[35801] {pull}36065[36065]
 - Allow parsing bytes in and bytes out as long integer in CEF processor. {issue}36100[36100] {pull}36108[36108]
+- Add support for registered owners and users to AzureAD entity analytics provider. {pull}36092[36092]
 
 *Auditbeat*
    - Migration of system/package module storage from gob encoding to flatbuffer encoding in bolt db. {pull}34817[34817]

x-pack/filebeat/docs/inputs/input-entity-analytics.asciidoc

@@ -185,6 +185,7 @@ Example device document:
     },
     "azure_ad": {
         "accountEnabled": true,
+        "deviceId": "2fbbb8f9-ff67-4a21-b867-a344d18a4198",
         "displayName": "DESKTOP-LETW452G",
         "operatingSystem": "Windows",
         "operatingSystemVersion": "10.0.19043.1337",
@@ -202,13 +203,39 @@ Example device document:
         ]
     },
     "device": {
-        "id": "2fbbb8f9-ff67-4a21-b867-a344d18a4198",
+        "id": "adbbe40a-0627-4328-89f1-88cac84dbc7f",
         "group": [
             {
                 "id": "331676df-b8fd-4492-82ed-02b927f8dd80",
                 "name": "group1"
             }
         ]
+        "registered_owners": [
+            {
+                "id": "5ebc6a0f-05b7-4f42-9c8a-682bbc75d0fc",
+                "userPrincipalName": "[email protected]",
+                "mail": "[email protected]",
+                "displayName": "Example User",
+                "givenName": "Example",
+                "surname": "User",
+                "jobTitle": "Software Engineer",
+                "mobilePhone": "123-555-1000",
+                "businessPhones": ["123-555-0122"]
+            },
+        ],
+        "registered_users": [
+            {
+                "id": "5ebc6a0f-05b7-4f42-9c8a-682bbc75d0fc",
+                "userPrincipalName": "[email protected]",
+                "mail": "[email protected]",
+                "displayName": "Example User",
+                "givenName": "Example",
+                "surname": "User",
+                "jobTitle": "Software Engineer",
+                "mobilePhone": "123-555-1000",
+                "businessPhones": ["123-555-0122"]
+            },
+        ],
     },
     "labels": {
         "identity_source": "azure-1"

x-pack/filebeat/input/entityanalytics/provider/azuread/azure.go

@@ -491,6 +491,36 @@ func (p *azure) publishDevice(d *fetcher.Device, state *stateStore, inputID stri
 		_, _ = deviceDoc.Put("device.group", groups)
 	}
 
+	owners := make([]mapstr.M, 0, d.RegisteredOwners.Len())
+	d.RegisteredOwners.ForEach(func(userID uuid.UUID) {
+		u, ok := state.users[userID]
+		if !ok {
+			p.logger.Warnf("Unable to lookup registered owner %q for device %q", userID, d.ID)
+			return
+		}
+		m := u.Fields.Clone()
+		_, _ = m.Put("user.id", u.ID.String())
+		owners = append(owners, m)
+	})
+	if len(owners) != 0 {
+		_, _ = deviceDoc.Put("device.registered_owners", owners)
+	}
+
+	users := make([]mapstr.M, 0, d.RegisteredUsers.Len())
+	d.RegisteredUsers.ForEach(func(userID uuid.UUID) {
+		u, ok := state.users[userID]
+		if !ok {
+			p.logger.Warnf("Unable to lookup registered user %q for device %q", userID, d.ID)
+			return
+		}
+		m := u.Fields.Clone()
+		_, _ = m.Put("user.id", u.ID.String())
+		users = append(users, m)
+	})
+	if len(users) != 0 {
+		_, _ = deviceDoc.Put("device.registered_users", users)
+	}
+
 	event := beat.Event{
 		Timestamp: time.Now(),
 		Fields:    deviceDoc,

x-pack/filebeat/input/entityanalytics/provider/azuread/fetcher/device.go

@@ -11,26 +11,6 @@ import (
 	"github.com/elastic/elastic-agent-libs/mapstr"
 )
 
-// TODO: Implement fetchers for the registered owners and users
-// for devices. These will then be retained in the following fields
-// of Device.
-//
-// 	// A set of UUIDs for registered owners of this device.
-// 	RegisteredOwners collections.UUIDSet `json:"registeredOwners"`
-// 	// A set of UUIDs for registered users of this device.
-// 	RegisteredUsers collections.UUIDSet `json:"registeredUsers"`
-//
-// and the addition of the following lines to Merge
-//
-// 	other.RegisteredOwners.ForEach(func(elem uuid.UUID) {
-// 		d.RegisteredOwners.Add(elem)
-// 	})
-// 	other.RegisteredUsers.ForEach(func(elem uuid.UUID) {
-// 		d.RegisteredUsers.Add(elem)
-// 	})
-//
-// with associated test extensions.
-
 // Device represents a device identity asset.
 type Device struct {
 	// The ID (UUIDv4) of the device.
@@ -41,6 +21,10 @@ type Device struct {
 	MemberOf collections.UUIDSet `json:"memberOf"`
 	// A set of UUIDs which are groups this device is a transitive member of.
 	TransitiveMemberOf collections.UUIDSet `json:"transitiveMemberOf"`
+	// A set of UUIDs for registered owners of this device.
+	RegisteredOwners collections.UUIDSet `json:"registeredOwners"`
+	// A set of UUIDs for registered users of this device.
+	RegisteredUsers collections.UUIDSet `json:"registeredUsers"`
 	// Discovered indicates that this device was newly discovered. This does not
 	// necessarily imply the device was recently added in Azure Active Directory,
 	// but it does indicate that it's the first time the device has been seen by
@@ -68,5 +52,11 @@ func (d *Device) Merge(other *Device) {
 	other.TransitiveMemberOf.ForEach(func(elem uuid.UUID) {
 		d.TransitiveMemberOf.Add(elem)
 	})
+	other.RegisteredOwners.ForEach(func(elem uuid.UUID) {
+		d.RegisteredOwners.Add(elem)
+	})
+	other.RegisteredUsers.ForEach(func(elem uuid.UUID) {
+		d.RegisteredUsers.Add(elem)
+	})
 	d.Deleted = other.Deleted
 }

x-pack/filebeat/input/entityanalytics/provider/azuread/fetcher/device_test.go

@@ -32,6 +32,11 @@ func TestDevice_Merge(t *testing.T) {
 				},
 				MemberOf:           collections.NewUUIDSet(uuid.MustParse("fcda226a-c920-4d99-81bc-d2d691a6c212")),
 				TransitiveMemberOf: collections.NewUUIDSet(uuid.MustParse("ca777ad5-9abf-4c9b-be1f-c38c6ec28f28")),
+				RegisteredOwners:   collections.NewUUIDSet(uuid.MustParse("c59fbdb8-e442-46b1-8d72-c8ac0b78ec0a")),
+				RegisteredUsers: collections.NewUUIDSet(
+					uuid.MustParse("27cea005-7377-4175-b2ef-e9d64c977f4d"),
+					uuid.MustParse("c59fbdb8-e442-46b1-8d72-c8ac0b78ec0a"),
+				),
 			},
 			InOther: &Device{
 				ID: uuid.MustParse("187f924c-e867-477e-8d74-dd762d6379dd"),
@@ -40,6 +45,11 @@ func TestDevice_Merge(t *testing.T) {
 				},
 				MemberOf:           collections.NewUUIDSet(uuid.MustParse("a77e8cbb-27a5-49d3-9d5e-801997621f87")),
 				TransitiveMemberOf: collections.NewUUIDSet(uuid.MustParse("c550d32c-09b2-4851-b0f2-1bc431e26d01")),
+				RegisteredOwners:   collections.NewUUIDSet(uuid.MustParse("81d1b5cd-7cd6-469d-9fe8-0a5c6cf2a7b6")),
+				RegisteredUsers: collections.NewUUIDSet(
+					uuid.MustParse("5e6d279a-ce2b-43b8-a38f-3110907e1974"),
+					uuid.MustParse("c59fbdb8-e442-46b1-8d72-c8ac0b78ec0a"),
+				),
 			},
 			Want: &Device{
 				ID: uuid.MustParse("187f924c-e867-477e-8d74-dd762d6379dd"),
@@ -55,6 +65,15 @@ func TestDevice_Merge(t *testing.T) {
 					uuid.MustParse("ca777ad5-9abf-4c9b-be1f-c38c6ec28f28"),
 					uuid.MustParse("c550d32c-09b2-4851-b0f2-1bc431e26d01"),
 				),
+				RegisteredOwners: collections.NewUUIDSet(
+					uuid.MustParse("81d1b5cd-7cd6-469d-9fe8-0a5c6cf2a7b6"),
+					uuid.MustParse("c59fbdb8-e442-46b1-8d72-c8ac0b78ec0a"),
+				),
+				RegisteredUsers: collections.NewUUIDSet(
+					uuid.MustParse("27cea005-7377-4175-b2ef-e9d64c977f4d"),
+					uuid.MustParse("5e6d279a-ce2b-43b8-a38f-3110907e1974"),
+					uuid.MustParse("c59fbdb8-e442-46b1-8d72-c8ac0b78ec0a"),
+				),
 			},
 		},
 	}
@@ -70,6 +89,8 @@ func TestDevice_Merge(t *testing.T) {
 			require.Equal(t, tc.Want.Fields, tc.In.Fields)
 			require.ElementsMatch(t, tc.Want.MemberOf.Values(), tc.In.MemberOf.Values(), "list A: Expected, listB: Actual")
 			require.ElementsMatch(t, tc.Want.TransitiveMemberOf.Values(), tc.In.TransitiveMemberOf.Values(), "list A: Expected, listB: Actual")
+			require.ElementsMatch(t, tc.Want.RegisteredOwners.Values(), tc.In.RegisteredOwners.Values(), "list A: Expected, listB: Actual")
+			require.ElementsMatch(t, tc.Want.RegisteredUsers.Values(), tc.In.RegisteredUsers.Values(), "list A: Expected, listB: Actual")
 		})
 	}
 }

x-pack/filebeat/input/entityanalytics/provider/azuread/fetcher/graph/graph.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/google/uuid"
 
+	"github.com/elastic/beats/v7/x-pack/filebeat/input/entityanalytics/internal/collections"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/entityanalytics/provider/azuread/authenticator"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/entityanalytics/provider/azuread/fetcher"
 	"github.com/elastic/elastic-agent-libs/config"
@@ -31,7 +32,7 @@ const (
 
 	defaultGroupsQuery  = "$select=displayName,members"
 	defaultUsersQuery   = "$select=accountEnabled,userPrincipalName,mail,displayName,givenName,surname,jobTitle,officeLocation,mobilePhone,businessPhones"
-	defaultDevicesQuery = "$select=accountEnabled,displayName,operatingSystem,operatingSystemVersion,physicalIds,extensionAttributes,alternativeSecurityIds"
+	defaultDevicesQuery = "$select=accountEnabled,deviceId,displayName,operatingSystem,operatingSystemVersion,physicalIds,extensionAttributes,alternativeSecurityIds"
 
 	apiGroupType  = "#microsoft.graph.group"
 	apiUserType   = "#microsoft.graph.user"
@@ -109,9 +110,10 @@ type graph struct {
 	logger *logp.Logger
 	auth   authenticator.Authenticator
 
-	usersURL   string
-	groupsURL  string
-	devicesURL string
+	usersURL           string
+	groupsURL          string
+	devicesURL         string
+	deviceOwnerUserURL string
 }
 
 // SetLogger sets the logger on this fetcher.
@@ -155,12 +157,12 @@ func (f *graph) Groups(ctx context.Context, deltaLink string) ([]*fetcher.Group,
 			return groups, response.DeltaLink, nil
 		}
 		if response.NextLink == fetchURL {
-			return nil, "", fmt.Errorf("error during fetch groups, encountered nextLink fetch infinite loop")
+			return groups, "", nextLinkLoopError{"groups"}
 		}
 		if response.NextLink != "" {
 			fetchURL = response.NextLink
 		} else {
-			return nil, "", fmt.Errorf("error during fetch groups, encountered response without nextLink or deltaLink")
+			return groups, "", missingLinkError{"groups"}
 		}
 	}
 }
@@ -207,12 +209,12 @@ func (f *graph) Users(ctx context.Context, deltaLink string) ([]*fetcher.User, s
 			return users, response.DeltaLink, nil
 		}
 		if response.NextLink == fetchURL {
-			return nil, "", fmt.Errorf("error during fetch users, encountered nextLink fetch infinite loop")
+			return users, "", nextLinkLoopError{"users"}
 		}
 		if response.NextLink != "" {
 			fetchURL = response.NextLink
 		} else {
-			return nil, "", fmt.Errorf("error during fetch users, encountered response without nextLink or deltaLink")
+			return users, "", missingLinkError{"users"}
 		}
 	}
 }
@@ -252,23 +254,41 @@ func (f *graph) Devices(ctx context.Context, deltaLink string) ([]*fetcher.Devic
 				continue
 			}
 			f.logger.Debugf("Got device %q from API", device.ID)
+
+			f.addRegistered(ctx, device, "registeredOwners", &device.RegisteredOwners)
+			f.addRegistered(ctx, device, "registeredUsers", &device.RegisteredUsers)
+
 			devices = append(devices, device)
 		}
 
 		if response.DeltaLink != "" {
 			return devices, response.DeltaLink, nil
 		}
 		if response.NextLink == fetchURL {
-			return nil, "", fmt.Errorf("error during fetch devices, encountered nextLink fetch infinite loop")
+			return devices, "", nextLinkLoopError{"devices"}
 		}
 		if response.NextLink != "" {
 			fetchURL = response.NextLink
 		} else {
-			return nil, "", fmt.Errorf("error during fetch devices, encountered response without nextLink or deltaLink")
+			return devices, "", missingLinkError{"devices"}
 		}
 	}
 }
 
+// addRegistered adds registered owner or user UUIDs to the provided device.
+func (f *graph) addRegistered(ctx context.Context, device *fetcher.Device, typ string, set *collections.UUIDSet) {
+	usersLink := fmt.Sprintf("%s/%s/%s", f.deviceOwnerUserURL, device.ID, typ) // ID here is the object ID.
+	users, _, err := f.Users(ctx, usersLink)
+	switch {
+	case err == nil, errors.Is(err, nextLinkLoopError{"users"}), errors.Is(err, missingLinkError{"users"}):
+	default:
+		f.logger.Errorf("Failed to obtain some registered user data: %w", err)
+	}
+	for _, u := range users {
+		set.Add(u.ID)
+	}
+}
+
 // doRequest is a convenience function for making HTTP requests to the Graph API.
 // It will automatically handle requesting a token using the authenticator attached
 // to this fetcher.
@@ -342,6 +362,15 @@ func New(cfg *config.C, logger *logp.Logger, auth authenticator.Authenticator) (
 	devicesURL.RawQuery = url.QueryEscape(defaultDevicesQuery)
 	f.devicesURL = devicesURL.String()
 
+	// The API takes a departure from the query approach here, so we
+	// need to construct a partial URL for use later when fetching
+	// registered owners and users.
+	ownerUserURL, err := url.Parse(f.conf.APIEndpoint + "/devices/")
+	if err != nil {
+		return nil, fmt.Errorf("invalid device owner/user URL endpoint: %w", err)
+	}
+	f.deviceOwnerUserURL = ownerUserURL.String()
+
 	return &f, nil
 }
 
@@ -423,3 +452,19 @@ func newDeviceFromAPI(d deviceAPI) (*fetcher.Device, error) {
 
 	return &newDevice, nil
 }
+
+type nextLinkLoopError struct {
+	endpoint string
+}
+
+func (e nextLinkLoopError) Error() string {
+	return fmt.Sprintf("error during fetch %s, encountered nextLink fetch infinite loop", e.endpoint)
+}
+
+type missingLinkError struct {
+	endpoint string
+}
+
+func (e missingLinkError) Error() string {
+	return fmt.Sprintf("error during fetch %s, encountered response without nextLink or deltaLink", e.endpoint)
+}