x-pack/filebeat/input/entityanalytics/provider/okta: add user group membership support

CHANGELOG.next.asciidoc

@@ -289,6 +289,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Make HTTP Endpoint input GA. {issue}38979[38979] {pull}39410[39410]
 - Update CEL mito extensions to v1.12.2. {pull}39755[39755]
 - Add support for base64-encoded HMAC headers to HTTP Endpoint. {pull}39655[39655]
+- Add user group membership support to Okta entity analytics provider. {issue}39814[39814] {pull}39815[39815]
 
 *Auditbeat*
 

x-pack/filebeat/input/entityanalytics/provider/okta/internal/okta/okta.go

@@ -9,6 +9,7 @@
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -98,6 +99,14 @@
 	Name *string `json:"name,omitempty"`
 }
 
+// Group is an Okta user group.
+//
+// See https://developer.okta.com/docs/reference/api/users/#request-parameters-8 (no anchor exists on the page for this endpoint) for details.
+type Group struct {
+	ID      string         `json:"id"`
+	Profile map[string]any `json:"profile"`
+}
+
 // Device is an Okta device's details.
 //
 // See https://developer.okta.com/docs/api/openapi/okta-management/management/tag/Device/#tag/Device/operation/listDevices for details
@@ -223,6 +232,27 @@
 	return getDetails[User](ctx, cli, u, key, user == "", omit, lim, window)
 }
 
+// GetUserGroupDetails returns Okta group details using the users API endpoint. host is the
+// Okta user domain and key is the API token to use for the query. user must not be empty.
+//
+// See GetUserDetails for details of the query and rate limit parameters.
+//
+// See https://developer.okta.com/docs/reference/api/users/#request-parameters-8 (no anchor exists on the page for this endpoint) for details.
+func GetUserGroupDetails(ctx context.Context, cli *http.Client, host, key, user string, lim *rate.Limiter, window time.Duration) ([]Group, http.Header, error) {
+	const endpoint = "/api/v1/users"
+
+	if user == "" {
+		return nil, nil, errors.New("no user specified")
+	}
+
+	u := &url.URL{
+		Scheme: "https",
+		Host:   host,
+		Path:   path.Join(endpoint, user, "groups"),
+	}
+	return getDetails[Group](ctx, cli, u, key, true, OmitNone, lim, window)
+}
+
 // GetDeviceDetails returns Okta device details using the list devices API endpoint. host is the
 // Okta user domain and key is the API token to use for the query. If device is not empty,
 // details for the specific device are returned, otherwise a list of all devices is returned.
@@ -242,7 +272,7 @@
 	return getDetails[Device](ctx, cli, u, key, device == "", OmitNone, lim, window)
 }
 
-// GetDeviceUsers returns Okta user details for users asscoiated with the provided device identifier
+// GetDeviceUsers returns Okta user details for users associated with the provided device identifier
 // using the list device users API. host is the Okta user domain and key is the API token to use for
 // the query. If device is empty, a nil User slice and header is returned, without error.
 //
@@ -276,7 +306,7 @@
 
 // entity is an Okta entity analytics entity.
 type entity interface {
-	User | Device | devUser
+	User | Group | Device | devUser
 }
 
 type devUser struct {
@@ -312,7 +342,7 @@
 	defer resp.Body.Close()
 	err = oktaRateLimit(resp.Header, window, lim)
 	if err != nil {
 		io.Copy(io.Discard, resp.Body)
 		return nil, nil, err
 	}
 

x-pack/filebeat/input/entityanalytics/provider/okta/internal/okta/okta_test.go

@@ -86,6 +86,31 @@
 				return
 			}
 
+			t.Run("my_groups", func(t *testing.T) {
+				query := make(url.Values)
+				query.Set("limit", "200")
+				groups, _, err := GetUserGroupDetails(context.Background(), http.DefaultClient, host, key, me.ID, limiter, window)
+				if err != nil {
+					t.Fatalf("unexpected error: %v", err)
+				}
+				if len(groups) == 0 {
+					t.Fatalf("unexpected len(groups): got:%d want:1", len(groups))
+				}
+
+				if omit&OmitCredentials != 0 && me.Credentials != nil {
+					t.Errorf("unexpected credentials with %s: %#v", omit, me.Credentials)
+				}
+
+				if !*logResponses {
+					return
+				}
+				b, err := json.Marshal(groups)
+				if err != nil {
+					t.Errorf("failed to marshal groups for logging: %v", err)
+				}
+				t.Logf("groups: %s", b)
+			})
+
 			t.Run("user", func(t *testing.T) {
 				if me.Profile.Login == "" {
 					b, _ := json.Marshal(me)
@@ -353,7 +378,7 @@
 func TestNext(t *testing.T) {
 	for i, test := range nextTests {
 		got, err := Next(test.header)
 		if err != test.wantErr {
 			t.Errorf("unexpected ok result for %d: got:%v want:%v", i, err, test.wantErr)
 		}
 		if got.Encode() != test.want {

x-pack/filebeat/input/entityanalytics/provider/okta/okta.go

@@ -385,15 +385,16 @@
 
 		if fullSync {
 			for _, u := range batch {
-				state.storeUser(u)
+				p.addGroup(ctx, u, state)
 				if u.LastUpdated.After(lastUpdated) {
 					lastUpdated = u.LastUpdated
 				}
 			}
 		} else {
 			users = grow(users, len(batch))
 			for _, u := range batch {
-				users = append(users, state.storeUser(u))
+				su := p.addGroup(ctx, u, state)
+				users = append(users, su)
 				if u.LastUpdated.After(lastUpdated) {
 					lastUpdated = u.LastUpdated
 				}
@@ -402,7 +403,7 @@
 
 		next, err := okta.Next(h)
 		if err != nil {
 			if err == io.EOF {
 				break
 			}
 			p.logger.Debugf("received %d users from API", len(users))
@@ -424,6 +425,17 @@
 	return users, nil
 }
 
+func (p *oktaInput) addGroup(ctx context.Context, u okta.User, state *stateStore) *User {
+	su := state.storeUser(u)
+	groups, _, err := okta.GetUserGroupDetails(ctx, p.client, p.cfg.OktaDomain, p.cfg.OktaToken, u.ID, p.lim, p.cfg.LimitWindow)
+	if err != nil {
+		p.logger.Warnf("failed to get user group membership for %s: %v", u.ID, err)
+		return su
+	}
+	su.Groups = groups
+	return su
+}
+
 // doFetchDevices handles fetching device and associated user identities from Okta.
 // If fullSync is true, then any existing deltaLink will be ignored, forcing a full
 // synchronization from Okta.
@@ -501,7 +513,7 @@
 
 				next, err := okta.Next(h)
 				if err != nil {
 					if err == io.EOF {
 						break
 					}
 					p.logger.Debugf("received %d devices from API", len(devices))
@@ -530,7 +542,7 @@
 
 		next, err := okta.Next(h)
 		if err != nil {
 			if err == io.EOF {
 				break
 			}
 			p.logger.Debugf("received %d devices from API", len(devices))

x-pack/filebeat/input/entityanalytics/provider/okta/okta_test.go

@@ -18,6 +18,7 @@ import (
 
 	"golang.org/x/time/rate"
 
+	"github.com/elastic/beats/v7/x-pack/filebeat/input/entityanalytics/provider/okta/internal/okta"
 	"github.com/elastic/elastic-agent-libs/logp"
 )
 
@@ -49,11 +50,13 @@ func TestOktaDoFetch(t *testing.T) {
 				window  = time.Minute
 				key     = "token"
 				users   = `[{"id":"USERID","status":"STATUS","created":"2023-05-14T13:37:20.000Z","activated":null,"statusChanged":"2023-05-15T01:50:30.000Z","lastLogin":"2023-05-15T01:59:20.000Z","lastUpdated":"2023-05-15T01:50:32.000Z","passwordChanged":"2023-05-15T01:50:32.000Z","type":{"id":"typeid"},"profile":{"firstName":"name","lastName":"surname","mobilePhone":null,"secondEmail":null,"login":"[email protected]","email":"[email protected]"},"credentials":{"password":{"value":"secret"},"emails":[{"value":"[email protected]","status":"VERIFIED","type":"PRIMARY"}],"provider":{"type":"OKTA","name":"OKTA"}},"_links":{"self":{"href":"https://localhost/api/v1/users/USERID"}}}]`
+				groups  = `[{"id":"USERID","profile":{"description":"All users in your organization","name":"Everyone"}}]`
 				devices = `[{"id":"DEVICEID","status":"STATUS","created":"2019-10-02T18:03:07.000Z","lastUpdated":"2019-10-02T18:03:07.000Z","profile":{"displayName":"Example Device name 1","platform":"WINDOWS","serialNumber":"XXDDRFCFRGF3M8MD6D","sid":"S-1-11-111","registered":true,"secureHardwarePresent":false,"diskEncryptionType":"ALL_INTERNAL_VOLUMES"},"resourceType":"UDDevice","resourceDisplayName":{"value":"Example Device name 1","sensitive":false},"resourceAlternateId":null,"resourceId":"DEVICEID","_links":{"activate":{"href":"https://localhost/api/v1/devices/DEVICEID/lifecycle/activate","hints":{"allow":["POST"]}},"self":{"href":"https://localhost/api/v1/devices/DEVICEID","hints":{"allow":["GET","PATCH","PUT"]}},"users":{"href":"https://localhost/api/v1/devices/DEVICEID/users","hints":{"allow":["GET"]}}}}]`
 			)
 
 			data := map[string]string{
 				"users":   users,
+				"groups":  groups,
 				"devices": devices,
 			}
 
@@ -63,6 +66,14 @@ func TestOktaDoFetch(t *testing.T) {
 				if err != nil {
 					t.Fatalf("failed to unmarshal user data: %v", err)
 				}
+				var wantGroups []okta.Group
+				err = json.Unmarshal([]byte(groups), &wantGroups)
+				if err != nil {
+					t.Fatalf("failed to unmarshal user data: %v", err)
+				}
+				for i, u := range wantUsers {
+					wantUsers[i].Groups = append(u.Groups, wantGroups...)
+				}
 			}
 			var wantDevices []Device
 			if test.wantDevices {
@@ -83,6 +94,12 @@ func TestOktaDoFetch(t *testing.T) {
 				w.Header().Add("x-rate-limit-remaining", "49")
 				w.Header().Add("x-rate-limit-reset", fmt.Sprint(time.Now().Add(time.Minute).Unix()))
 
+				if strings.HasPrefix(r.URL.Path, "/api/v1/users") && strings.HasSuffix(r.URL.Path, "groups") {
+					// Give the groups if this is a get user groups request.
+					userid := strings.TrimSuffix(strings.TrimPrefix(r.URL.Path, "/api/v1/users/"), "/groups")
+					fmt.Fprintln(w, strings.ReplaceAll(data["groups"], "USERID", userid))
+					return
+				}
 				if strings.HasPrefix(r.URL.Path, "/api/v1/device") && strings.HasSuffix(r.URL.Path, "users") {
 					// Give one user if this is a get device users request.
 					fmt.Fprintln(w, data["users"])
@@ -158,9 +175,15 @@ func TestOktaDoFetch(t *testing.T) {
 					t.Errorf("unexpected number of results: got:%d want:%d", len(got), wantCount(repeats, test.wantUsers))
 				}
 				for i, g := range got {
-					if wantID := fmt.Sprintf("userid%d", i+1); g.ID != wantID {
+					wantID := fmt.Sprintf("userid%d", i+1)
+					if g.ID != wantID {
 						t.Errorf("unexpected user ID for user %d: got:%s want:%s", i, g.ID, wantID)
 					}
+					for j, gg := range g.Groups {
+						if gg.ID != wantID {
+							t.Errorf("unexpected used ID for user group %d in %d: got:%s want:%s", j, i, gg.ID, wantID)
+						}
+					}
 					if g.State != wantStates[g.ID] {
 						t.Errorf("unexpected user state for user %s: got:%s want:%s", g.ID, g.State, wantStates[g.ID])
 					}

x-pack/filebeat/input/entityanalytics/provider/okta/statestore.go

@@ -39,7 +39,8 @@ const (
 
 type User struct {
 	okta.User `json:"properties"`
-	State     State `json:"state"`
+	Groups    []okta.Group `json:"groups"`
+	State     State        `json:"state"`
 }
 
 type Device struct {