From 14745f6b5749340f8d6befba53f27249b0a1fd3f Mon Sep 17 00:00:00 2001
From: Dan Kortschak <dan.kortschak@elastic.co>
Date: Fri, 6 Oct 2023 06:23:29 +1030
Subject: [PATCH] x-pack/filebeat/input/entityanalytics/provider/okta: avoid
 work on unwanted datasets

During full sync the provider may have state from a previous dataset. So
in the case that the user has changed dataset from users to devices or
vice versa the provider may publish already existing state in the entity
graph. This change adds conditional checks to ensure that unwanted
dataset records are not published.
---
 CHANGELOG.next.asciidoc                       |  1 +
 .../entityanalytics/provider/okta/conf.go     | 18 +++++++++++++
 .../entityanalytics/provider/okta/okta.go     | 25 ++++++++++---------
 3 files changed, 32 insertions(+), 12 deletions(-)

diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
index 6932839c03f2..7edf65335117 100644
--- a/CHANGELOG.next.asciidoc
+++ b/CHANGELOG.next.asciidoc
@@ -234,6 +234,7 @@ is collected by it.
 - Re-use buffers to optimise memory allocation in fingerprint mode of filestream {pull}36736[36736]
 - Allow http_endpoint input to receive PUT and PATCH requests. {pull}36734[36734]
 - Add cache processor. {pull}36786[36786]
+- Avoid unwanted publication of Okta entity records. {pull}36770[36770]
 
 *Auditbeat*
 
diff --git a/x-pack/filebeat/input/entityanalytics/provider/okta/conf.go b/x-pack/filebeat/input/entityanalytics/provider/okta/conf.go
index e344b56478fc..eb0906e78d54 100644
--- a/x-pack/filebeat/input/entityanalytics/provider/okta/conf.go
+++ b/x-pack/filebeat/input/entityanalytics/provider/okta/conf.go
@@ -168,3 +168,21 @@ func (c *conf) Validate() error {
 		return errors.New("dataset must be 'all', 'users', 'devices' or empty")
 	}
 }
+
+func (c *conf) wantUsers() bool {
+	switch strings.ToLower(c.Dataset) {
+	case "", "all", "users":
+		return true
+	default:
+		return false
+	}
+}
+
+func (c *conf) wantDevices() bool {
+	switch strings.ToLower(c.Dataset) {
+	case "", "all", "devices":
+		return true
+	default:
+		return false
+	}
+}
diff --git a/x-pack/filebeat/input/entityanalytics/provider/okta/okta.go b/x-pack/filebeat/input/entityanalytics/provider/okta/okta.go
index 4aff3cd3e595..d56ae7570600 100644
--- a/x-pack/filebeat/input/entityanalytics/provider/okta/okta.go
+++ b/x-pack/filebeat/input/entityanalytics/provider/okta/okta.go
@@ -12,7 +12,6 @@ import (
 	"io"
 	"net/http"
 	"net/url"
-	"strings"
 	"time"
 
 	"github.com/hashicorp/go-retryablehttp"
@@ -253,16 +252,22 @@ func (p *oktaInput) runFullSync(inputCtx v2.Context, store *kvstore.Store, clien
 		return err
 	}
 
-	if len(state.users) != 0 || len(state.devices) != 0 {
+	wantUsers := p.cfg.wantUsers()
+	wantDevices := p.cfg.wantDevices()
+	if (len(state.users) != 0 && wantUsers) || (len(state.devices) != 0 && wantDevices) {
 		tracker := kvstore.NewTxTracker(ctx)
 
 		start := time.Now()
 		p.publishMarker(start, start, inputCtx.ID, true, client, tracker)
-		for _, u := range state.users {
-			p.publishUser(u, state, inputCtx.ID, client, tracker)
+		if wantUsers {
+			for _, u := range state.users {
+				p.publishUser(u, state, inputCtx.ID, client, tracker)
+			}
 		}
-		for _, d := range state.devices {
-			p.publishDevice(d, state, inputCtx.ID, client, tracker)
+		if wantDevices {
+			for _, d := range state.devices {
+				p.publishDevice(d, state, inputCtx.ID, client, tracker)
+			}
 		}
 
 		end := time.Now()
@@ -339,9 +344,7 @@ func (p *oktaInput) runIncrementalUpdate(inputCtx v2.Context, store *kvstore.Sto
 // any existing deltaLink will be ignored, forcing a full synchronization from Okta.
 // Returns a set of modified users by ID.
 func (p *oktaInput) doFetchUsers(ctx context.Context, state *stateStore, fullSync bool) ([]*User, error) {
-	switch strings.ToLower(p.cfg.Dataset) {
-	case "", "all", "users":
-	default:
+	if !p.cfg.wantUsers() {
 		p.logger.Debugf("Skipping user collection from API: dataset=%s", p.cfg.Dataset)
 		return nil, nil
 	}
@@ -426,9 +429,7 @@ func (p *oktaInput) doFetchUsers(ctx context.Context, state *stateStore, fullSyn
 // synchronization from Okta.
 // Returns a set of modified devices by ID.
 func (p *oktaInput) doFetchDevices(ctx context.Context, state *stateStore, fullSync bool) ([]*Device, error) {
-	switch strings.ToLower(p.cfg.Dataset) {
-	case "", "all", "devices":
-	default:
+	if !p.cfg.wantDevices() {
 		p.logger.Debugf("Skipping device collection from API: dataset=%s", p.cfg.Dataset)
 		return nil, nil
 	}