[Filebeat][CEL] - Added support for Okta Oauth2 provider

CHANGELOG.next.asciidoc

@@ -216,6 +216,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Allow fine-grained control of entity analytics API requests for Okta provider. {issue}36440[36440] {pull}36492[36492]
 - Add support for expanding `journald.process.capabilities` into the human-readable effective capabilities in the ECS `process.thread.capabilities.effective` field. {issue}36454[36454] {pull}36470[36470]
 - Allow fine-grained control of entity analytics API requests for AzureAD provider. {issue}36440[36440] {pull}36441[36441]
+- Added support for Okta OAuth2 provider in the CEL input. {issue}36336[36336] {pull}36521[36521]
 
 *Auditbeat*
 

x-pack/filebeat/docs/inputs/input-cel.asciidoc

@@ -409,19 +409,19 @@ the `auth.oauth2` section is missing.
 
 Used to configure supported oauth2 providers.
 Each supported provider will require specific settings. It is not set by default.
-Supported providers are: `azure`, `google`.
+Supported providers are: `azure`, `google`, `okta`.
 
 [float]
 ==== `auth.oauth2.client.id`
 
 The client ID used as part of the authentication flow. It is always required
-except if using `google` as provider. Required for providers: `default`, `azure`.
+except if using `google` as provider. Required for providers: `default`, `azure`, `okta`.
 
 [float]
 ==== `auth.oauth2.client.secret`
 
 The client secret used as part of the authentication flow. It is always required
-except if using `google` as provider. Required for providers: `default`, `azure`.
+except if using `google` or `okta` as provider. Required for providers: `default`, `azure`.
 
 [float]
 ==== `auth.oauth2.user`
@@ -528,6 +528,20 @@ how to provide Google credentials, please refer to https://cloud.google.com/docs
 Email of the delegated account used to create the credentials (usually an admin). Used in combination
 with `auth.oauth2.google.jwt_file` or `auth.oauth2.google.jwt_json`.
 
+[float]
+==== `auth.oauth2.okta.jwk_file`
+
+The RSA JWK Private Key file for your Okta Service App which is used for interacting with Okta Org Auth Server to mint tokens with okta.* scopes.
+
+NOTE: Only one of the credentials settings can be set at once. For more information please refer to https://developer.okta.com/docs/guides/implement-oauth-for-okta-serviceapp/main/
+
+[float]
+==== `auth.oauth2.okta.jwk_json`
+
+The RSA JWK Private Key JSON for your Okta Service App which is used for interacting with Okta Org Auth Server to mint tokens with okta.* scopes.
+
+NOTE: Only one of the credentials settings can be set at once. For more information please refer to https://developer.okta.com/docs/guides/implement-oauth-for-okta-serviceapp/main/
+
 [[resource-parameters]]
 [float]
 ==== `resource.url`

x-pack/filebeat/input/cel/config_auth.go

@@ -66,6 +66,7 @@ const (
 	oAuth2ProviderDefault oAuth2Provider = ""       // oAuth2ProviderDefault means no specific provider is set.
 	oAuth2ProviderAzure   oAuth2Provider = "azure"  // oAuth2ProviderAzure AzureAD.
 	oAuth2ProviderGoogle  oAuth2Provider = "google" // oAuth2ProviderGoogle Google.
+	oAuth2ProviderOkta    oAuth2Provider = "okta"   // oAuth2ProviderOkta Okta.
 )
 
 func (p *oAuth2Provider) Unpack(in string) error {
@@ -100,6 +101,10 @@ type oAuth2Config struct {
 	// microsoft azure specific
 	AzureTenantID string `config:"azure.tenant_id"`
 	AzureResource string `config:"azure.resource"`
+
+	// okta specific RSA JWK private key
+	OktaJWKFile string          `config:"okta.jwk_file"`
+	OktaJWKJSON common.JSONBlob `config:"okta.jwk_json"`
 }
 
 // IsEnabled returns true if the `enable` field is set to true in the yaml.
@@ -160,6 +165,8 @@ func (o *oAuth2Config) client(ctx context.Context, client *http.Client) (*http.C
 			return nil, fmt.Errorf("oauth2 client: error loading credentials: %w", err)
 		}
 		return oauth2.NewClient(ctx, creds.TokenSource), nil
+	case oAuth2ProviderOkta:
+		return o.fetchOktaOauthClient(ctx, client)
 	default:
 		return nil, errors.New("oauth2 client: unknown provider")
 	}
@@ -216,6 +223,8 @@ func (o *oAuth2Config) Validate() error {
 		return o.validateAzureProvider()
 	case oAuth2ProviderGoogle:
 		return o.validateGoogleProvider()
+	case oAuth2ProviderOkta:
+		return o.validateOktaProvider()
 	case oAuth2ProviderDefault:
 		if o.TokenURL == "" || o.ClientID == "" || o.ClientSecret == nil {
 			return errors.New("both token_url and client credentials must be provided")
@@ -275,6 +284,22 @@ func (o *oAuth2Config) validateGoogleProvider() error {
 	return fmt.Errorf("no authentication credentials were configured or detected (ADC)")
 }
 
+func (o *oAuth2Config) validateOktaProvider() error {
+	if o.TokenURL == "" || o.ClientID == "" || len(o.Scopes) == 0 || (o.OktaJWKJSON == nil && o.OktaJWKFile == "") {
+		return errors.New("okta validation error: token_url, client_id, scopes and at least one of okta.jwk_json or okta.jwk_file must be provided")
+	}
+	// jwk_file
+	if o.OktaJWKFile != "" {
+		return populateJSONFromFile(o.OktaJWKFile, &o.OktaJWKJSON)
+	}
+	// jwk_json
+	if len(o.OktaJWKJSON) != 0 {
+		return nil
+	}
+
+	return fmt.Errorf("okta validation error: no authentication credentials were configured or detected")
+}
+
 func populateJSONFromFile(file string, dst *common.JSONBlob) error {
 	if _, err := os.Stat(file); errors.Is(err, fs.ErrNotExist) {
 		return fmt.Errorf("the file %q cannot be found", file)

x-pack/filebeat/input/cel/config_okta_auth.go

@@ -0,0 +1,190 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package cel
+
+import (
+	"context"
+	"crypto/rsa"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"math/big"
+	"net/http"
+	"net/url"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/lestrrat-go/jwx/v2/jwa"
+	"github.com/lestrrat-go/jwx/v2/jwt"
+	"golang.org/x/oauth2"
+	"golang.org/x/oauth2/clientcredentials"
+)
+
+// oktaTokenSource is a custom implementation of the oauth2.TokenSource interface.
+// For more information, see https://pkg.go.dev/golang.org/x/oauth2#TokenSource
+type oktaTokenSource struct {
+	mu      sync.Mutex
+	ctx     context.Context
+	conf    *oauth2.Config
+	token   *oauth2.Token
+	oktaJWK []byte
+}
+
+// fetchOktaOauthClient fetches an OAuth2 client using the Okta JWK credentials.
+func (o *oAuth2Config) fetchOktaOauthClient(ctx context.Context, _ *http.Client) (*http.Client, error) {
+	conf := &oauth2.Config{
+		ClientID: o.ClientID,
+		Scopes:   o.Scopes,
+		Endpoint: oauth2.Endpoint{
+			TokenURL: o.TokenURL,
+		},
+	}
+
+	oktaJWT, err := generateOktaJWT(o.OktaJWKJSON, conf)
+	if err != nil {
+		return nil, fmt.Errorf("oauth2 client: error generating Okta JWT: %w", err)
+	}
+
+	token, err := exchangeForBearerToken(ctx, oktaJWT, conf)
+	if err != nil {
+		return nil, fmt.Errorf("oauth2 client: error exchanging Okta JWT for bearer token: %w", err)
+	}
+
+	tokenSource := &oktaTokenSource{
+		conf:    conf,
+		ctx:     ctx,
+		oktaJWK: o.OktaJWKJSON,
+		token:   token,
+	}
+	// reuse the tokenSource to refresh the token (automatically calls the custom Token() method when token is no longer valid).
+	client := oauth2.NewClient(ctx, oauth2.ReuseTokenSource(token, tokenSource))
+
+	return client, nil
+}
+
+// Token implements the oauth2.TokenSource interface and helps to implement custom token refresh logic.
+// Parent context is passed via the customTokenSource struct since we cannot modify the function signature here.
+func (ts *oktaTokenSource) Token() (*oauth2.Token, error) {
+	ts.mu.Lock()
+	defer ts.mu.Unlock()
+
+	oktaJWT, err := generateOktaJWT(ts.oktaJWK, ts.conf)
+	if err != nil {
+		return nil, fmt.Errorf("error generating Okta JWT: %w", err)
+	}
+	token, err := exchangeForBearerToken(ts.ctx, oktaJWT, ts.conf)
+	if err != nil {
+		return nil, fmt.Errorf("error exchanging Okta JWT for bearer token: %w", err)
+
+	}
+
+	return token, nil
+}
+
+func generateOktaJWT(oktaJWK []byte, cnf *oauth2.Config) (string, error) {
+	// unmarshal the JWK into a map
+	var jwkData map[string]string
+	err := json.Unmarshal(oktaJWK, &jwkData)
+	if err != nil {
+		return "", fmt.Errorf("error decoding JWK: %w", err)
+	}
+
+	// create an RSA private key from JWK components
+	decodeBase64 := func(key string) (*big.Int, error) {
+		data, err := base64.RawURLEncoding.DecodeString(jwkData[key])
+		if err != nil {
+			return nil, fmt.Errorf("error decoding RSA JWK component %s: %w", key, err)
+		}
+		return new(big.Int).SetBytes(data), nil
+	}
+
+	n, err := decodeBase64("n")
+	if err != nil {
+		return "", err
+	}
+	e, err := decodeBase64("e")
+	if err != nil {
+		return "", err
+	}
+	d, err := decodeBase64("d")
+	if err != nil {
+		return "", err
+	}
+	p, err := decodeBase64("p")
+	if err != nil {
+		return "", err
+	}
+	q, err := decodeBase64("q")
+	if err != nil {
+		return "", err
+	}
+	dp, err := decodeBase64("dp")
+	if err != nil {
+		return "", err
+	}
+	dq, err := decodeBase64("dq")
+	if err != nil {
+		return "", err
+	}
+	qi, err := decodeBase64("qi")
+	if err != nil {
+		return "", err
+	}
+
+	privateKeyRSA := &rsa.PrivateKey{
+		PublicKey: rsa.PublicKey{
+			N: n,
+			E: int(e.Int64()),
+		},
+		D:      d,
+		Primes: []*big.Int{p, q},
+		Precomputed: rsa.PrecomputedValues{
+			Dp:   dp,
+			Dq:   dq,
+			Qinv: qi,
+		},
+	}
+
+	// create a JWT token using required claims and sign it with the private key
+	now := time.Now()
+	tok, err := jwt.NewBuilder().Audience([]string{cnf.Endpoint.TokenURL}).
+		Issuer(cnf.ClientID).
+		Subject(cnf.ClientID).
+		IssuedAt(now).
+		Expiration(now.Add(time.Hour)).
+		Build()
+	if err != nil {
+		return "", err
+	}
+	signedToken, err := jwt.Sign(tok, jwt.WithKey(jwa.RS256, privateKeyRSA))
+	if err != nil {
+		return "", fmt.Errorf("failed to sign token: %w", err)
+	}
+
+	return string(signedToken), nil
+}
+
+// exchangeForBearerToken exchanges the Okta JWT for a bearer token.
+func exchangeForBearerToken(ctx context.Context, oktaJWT string, cnf *oauth2.Config) (*oauth2.Token, error) {
+	data := url.Values{}
+	data.Set("grant_type", "client_credentials")
+	data.Set("scope", strings.Join(cnf.Scopes, " "))
+	data.Set("client_assertion_type", "urn:ietf:params:oauth:client-assertion-type:jwt-bearer")
+	data.Set("client_assertion", oktaJWT)
+	oauthConfig := &clientcredentials.Config{
+		TokenURL:       cnf.Endpoint.TokenURL,
+		EndpointParams: data,
+	}
+	tokenSource := oauthConfig.TokenSource(ctx)
+
+	// get the access token
+	accessToken, err := tokenSource.Token()
+	if err != nil {
+		return nil, err
+	}
+
+	return accessToken, nil
+}

x-pack/filebeat/input/cel/config_test.go

@@ -444,6 +444,43 @@ var oAuth2ValidationTests = []struct {
 			},
 		},
 	},
+	{
+		name:    "okta requires token_url, client_id, scopes and at least one of okta.jwk_json or okta.jwk_file to be provided",
+		wantErr: errors.New("okta validation error: token_url, client_id, scopes and at least one of okta.jwk_json or okta.jwk_file must be provided accessing 'auth.oauth2'"),
+		input: map[string]interface{}{
+			"auth.oauth2": map[string]interface{}{
+				"provider":  "okta",
+				"client.id": "a_client_id",
+				"token_url": "localhost",
+				"scopes":    []string{"foo"},
+			},
+		},
+	},
+	{
+		name:    "okta oauth2 validation fails if jwk_json is not a valid JSON",
+		wantErr: errors.New("the field can't be converted to valid JSON accessing 'auth.oauth2.okta.jwk_json'"),
+		input: map[string]interface{}{
+			"auth.oauth2": map[string]interface{}{
+				"provider":      "okta",
+				"client.id":     "a_client_id",
+				"token_url":     "localhost",
+				"scopes":        []string{"foo"},
+				"okta.jwk_json": `"p":"x","kty":"RSA","q":"x","d":"x","e":"x","use":"x","kid":"x","qi":"x","dp":"x","alg":"x","dq":"x","n":"x"}`,
+			},
+		},
+	},
+	{
+		name: "okta successful oauth2 validation",
+		input: map[string]interface{}{
+			"auth.oauth2": map[string]interface{}{
+				"provider":      "okta",
+				"client.id":     "a_client_id",
+				"token_url":     "localhost",
+				"scopes":        []string{"foo"},
+				"okta.jwk_json": `{"p":"x","kty":"RSA","q":"x","d":"x","e":"x","use":"x","kid":"x","qi":"x","dp":"x","alg":"x","dq":"x","n":"x"}`,
+			},
+		},
+	},
 }
 
 func TestConfigOauth2Validation(t *testing.T) {