x-pack/filebeat/input/cel: add support for pem encoded keys

This adds a new Okta auth field, jwk_pem, that allows users to specify a PEM-encoded
private key for authentication.

Also refactor the JSON-based code to simplify and add minimal testing.

CHANGELOG.next.asciidoc

@@ -178,6 +178,7 @@ Setting environmental variable ELASTIC_NETINFO:false in Elastic Agent pod will d
 - Add request trace logging for chained API requests. {issue}37551[36551] {pull}37682[37682]
 - Relax TCP/UDP metric polling expectations to improve metric collection. {pull}37714[37714]
 - Add support for PEM-based Okta auth in HTTPJSON. {pull}37772[37772]
+- Add support for PEM-based Okta auth in CEL. {pull}37813[37813]
 
 *Auditbeat*
 

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

@@ -580,6 +580,13 @@ The RSA JWK Private Key JSON for your Okta Service App which is used for interac
 
 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_pem`
+
+The RSA JWK private key PEM block 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

@@ -6,6 +6,7 @@ package cel
 
 import (
 	"context"
+	"crypto/x509"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -141,6 +142,7 @@ type oAuth2Config struct {
 	// okta specific RSA JWK private key
 	OktaJWKFile string          `config:"okta.jwk_file"`
 	OktaJWKJSON common.JSONBlob `config:"okta.jwk_json"`
+	OktaJWKPEM  string          `config:"okta.jwk_pem"`
 }
 
 // isEnabled returns true if the `enable` field is set to true in the yaml.
@@ -321,8 +323,26 @@ func (o *oAuth2Config) validateGoogleProvider() error {
 }
 
 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")
+	if o.TokenURL == "" || o.ClientID == "" || len(o.Scopes) == 0 {
+		return errors.New("okta validation error: token_url, client_id, scopes must be provided")
+	}
+	var n int
+	if o.OktaJWKJSON != nil {
+		n++
+	}
+	if o.OktaJWKFile != "" {
+		n++
+	}
+	if o.OktaJWKPEM != "" {
+		n++
+	}
+	if n != 1 {
+		return errors.New("okta validation error: one of okta.jwk_json, okta.jwk_file or okta.jwk_pem must be provided")
+	}
+	// jwk_pem
+	if o.OktaJWKPEM != "" {
+		_, err := x509.ParsePKCS1PrivateKey([]byte(o.OktaJWKPEM))
+		return err
 	}
 	// jwk_file
 	if o.OktaJWKFile != "" {

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

@@ -5,10 +5,13 @@
 package cel
 
 import (
+	"bytes"
 	"context"
 	"crypto/rsa"
+	"crypto/x509"
 	"encoding/base64"
 	"encoding/json"
+	"encoding/pem"
 	"fmt"
 	"math/big"
 	"net/http"
@@ -43,9 +46,20 @@ func (o *oAuth2Config) fetchOktaOauthClient(ctx context.Context, _ *http.Client)
 		},
 	}
 
-	oktaJWT, err := generateOktaJWT(o.OktaJWKJSON, conf)
-	if err != nil {
-		return nil, fmt.Errorf("oauth2 client: error generating Okta JWT: %w", err)
+	var (
+		oktaJWT string
+		err     error
+	)
+	if len(o.OktaJWKPEM) != 0 {
+		oktaJWT, err = generateOktaJWTPEM(o.OktaJWKPEM, conf)
+		if err != nil {
+			return nil, fmt.Errorf("oauth2 client: error generating Okta JWT PEM: %w", err)
+		}
+	} else {
+		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)
@@ -59,14 +73,16 @@ func (o *oAuth2Config) fetchOktaOauthClient(ctx context.Context, _ *http.Client)
 		oktaJWK: o.OktaJWKJSON,
 		token:   token,
 	}
-	// reuse the tokenSource to refresh the token (automatically calls the custom Token() method when token is no longer valid).
+	// 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.
+// Token implements the oauth2.TokenSource interface and helps to implement
+// custom token refresh logic. The 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()
@@ -85,70 +101,79 @@ func (ts *oktaTokenSource) Token() (*oauth2.Token, error) {
 }
 
 func generateOktaJWT(oktaJWK []byte, cnf *oauth2.Config) (string, error) {
-	// unmarshal the JWK into a map
-	var jwkData map[string]string
+	// Unmarshal the JWK into big ints.
+	var jwkData struct {
+		N    base64int `json:"n"`
+		E    base64int `json:"e"`
+		D    base64int `json:"d"`
+		P    base64int `json:"p"`
+		Q    base64int `json:"q"`
+		Dp   base64int `json:"dp"`
+		Dq   base64int `json:"dq"`
+		Qinv base64int `json:"qi"`
+	}
 	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
+	// Create an RSA private key from JWK components.
+	key := &rsa.PrivateKey{
+		PublicKey: rsa.PublicKey{
+			N: &jwkData.N.Int,
+			E: int(jwkData.E.Int64()),
+		},
+		D:      &jwkData.D.Int,
+		Primes: []*big.Int{&jwkData.P.Int, &jwkData.Q.Int},
+		Precomputed: rsa.PrecomputedValues{
+			Dp:   &jwkData.Dp.Int,
+			Dq:   &jwkData.Dq.Int,
+			Qinv: &jwkData.Qinv.Int,
+		},
 	}
 
-	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
+	return signJWT(cnf, key)
+
+}
+
+// base64int is a JSON decoding shim for base64-encoded big.Int.
+type base64int struct {
+	big.Int
+}
+
+func (i *base64int) UnmarshalJSON(b []byte) error {
+	src, ok := bytes.CutPrefix(b, []byte{'"'})
+	if !ok {
+		return fmt.Errorf("invalid JSON type: %s", b)
 	}
-	q, err := decodeBase64("q")
-	if err != nil {
-		return "", err
+	src, ok = bytes.CutSuffix(src, []byte{'"'})
+	if !ok {
+		return fmt.Errorf("invalid JSON type: %s", b)
 	}
-	dp, err := decodeBase64("dp")
+	dst := make([]byte, base64.RawURLEncoding.DecodedLen(len(src)))
+	_, err := base64.RawURLEncoding.Decode(dst, src)
 	if err != nil {
-		return "", err
+		return err
 	}
-	dq, err := decodeBase64("dq")
-	if err != nil {
-		return "", err
+	i.SetBytes(dst)
+	return nil
+}
+
+func generateOktaJWTPEM(pemdata string, cnf *oauth2.Config) (string, error) {
+	blk, rest := pem.Decode([]byte(pemdata))
+	if rest := bytes.TrimSpace(rest); len(rest) != 0 {
+		return "", fmt.Errorf("PEM text has trailing data: %s", rest)
 	}
-	qi, err := decodeBase64("qi")
+	key, err := x509.ParsePKCS8PrivateKey(blk.Bytes)
 	if err != nil {
 		return "", err
 	}
+	return signJWT(cnf, key)
+}
 
-	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
+// signJWT creates a JWT token using required claims and sign it with the
+// private key.
+func signJWT(cnf *oauth2.Config, key any) (string, error) {
 	now := time.Now()
 	tok, err := jwt.NewBuilder().Audience([]string{cnf.Endpoint.TokenURL}).
 		Issuer(cnf.ClientID).
@@ -159,11 +184,10 @@ func generateOktaJWT(oktaJWK []byte, cnf *oauth2.Config) (string, error) {
 	if err != nil {
 		return "", err
 	}
-	signedToken, err := jwt.Sign(tok, jwt.WithKey(jwa.RS256, privateKeyRSA))
+	signedToken, err := jwt.Sign(tok, jwt.WithKey(jwa.RS256, key))
 	if err != nil {
 		return "", fmt.Errorf("failed to sign token: %w", err)
 	}
-
 	return string(signedToken), nil
 }
 

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

@@ -0,0 +1,88 @@
+// 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 (
+	"testing"
+
+	"github.com/lestrrat-go/jwx/v2/jwt"
+	"golang.org/x/oauth2"
+)
+
+func TestGenerateOktaJWT(t *testing.T) {
+	// jwt is a JWT obtained from the Okta integration.
+	const jwtText = `{ "d": "Cmhokw2MnZfX6da36nnsnQ7IPX9vE6se8_D1NgyL9j9rarYpexhlp45hswcAIFNgWA03NV848Gc0e84AW6wMbyD2E8LPI0Bd8lhdmzRE6L4or2Rxqqjk2Pr2aqGnqs4A0uTijAA7MfPF1zFFdR3EOVx499fEeTiMcLjO83IJCoNiOySDoQgt3KofX5bCbaDy2eiB83rzf0fEcWrWfTY65_Hc2c5lek-1uuF7NpELVzX80p5H-b9MOfLn0BdOGe-mJ2j5bXi-UCQ45Wxj2jdkoA_Qwb4MEtXZjp5LjcM75SrlGfVd99acML2wGZgYLGweJ0sAPDlKzGvj4ve-JT8nNw", "p": "8-UBb4psN0wRPktkh3S48L3ng4T5zR08t7nwXDYNajROrS2j7oq60dtlGY4IwgwcC0c9GDQP7NiN2IpU2uahYkGQ7lDyM_h7UfQWL5fMrsYiKgn2pUgSy5TTT8smkSLbJAD35nAH6PknsQ2PuvOlb4laiC0MXw1Rw4vT9HAEB9M", "q": "0DJkPEN0bECG_6lorlNJgIfoNahVevGKK-Yti1YZ5K-nQCuffPCwPG0oZZo_55y5LODe9W7psxnAt7wxkpAY4lK2hpHTWJSkPjqXWFYIP8trn4RZDShnJXli0i1XqPOqkiVzBZGx5nLtj2bUtmXfIU7-kneHGvLQ5EXcyQW1ISM", "dp": "Ye1PWEPSE5ndSo_m-2RoZXE6pdocmrjkijiEQ-IIHN6HwI0Ux1C4lk5rF4mqBo_qKrUd2Lv-sPB6c7mHPKVhoxwEX0vtE-TvTwacadufeYVgblS1zcNUmJ1XAzDkeV3vc1NYNhRBeM-hmjuBvGTbxh72VLsRvpCQhd186yaW17U", "dq": "jvSK7vZCUrJb_-CLCGgX6DFpuK5FQ43mmg4K58nPLb-Oz_kkId4CpPsu6dToXFi4raAad9wYi-n68i4-u6xF6eFxgyVOQVyPCkug7_7i2ysKUxXFL8u2R3z55edMca4eSQt91y0bQmlXxUeOd0-rzms3UcrQ8igYVyXBXCaXIJE", "qi": "iIY1Y4bzMYIFG7XH7gNP7C-mWi6QH4l9aGRTzPB_gPaFThvc0XKW0S0l82bfp_PPPWg4D4QpDCp7rZ6KhEA8BlNi86Vt3V6F3Hz5XiDa4ikgQNsAXiXLqf83R-y1-cwHjW70PP3U89hmalCRRFfVXcLHV77AVHqbrp9rAIo-X-I", "kty": "RSA", "e": "AQAB", "kid": "koeFQjkyiav_3Qwr3aRinCqCD2LaEHOjFnje7XlkbdI", "n": "xloTY8bAuI5AEo8JursCd7w0LmELCae7JOFaVo9njGrG8tRNqgIdjPyoGY_ABwKkmjcCMLGMA29llFDbry8rB4LTWai-h_jX4_uUUnl52mLX-lO6merL5HEPZF438Ql9Hrxs5yGzT8n865-E_3uwYSBrhTjvlZJeXYUeVHfKo8pJSSsw3RZEjBW4Tt0eFmCZnFErtTyk3oUPaYVP-8YLLAenhUDV4Lm1dC4dxqUj0Oh6XrWgIb-eYHGolMY9g9xbgyd4ir39RodA_1DOjzHWpNfCM-J5ZOtfpuKCAe5__u7L8FT0m56XOxcDoVVsz1J1VNrACWAGbhDWNjyHfL5E2Q" }`
+	cnf := &oauth2.Config{
+		ClientID: "0oaajljpeokFZLyKU5d7",
+		Scopes:   []string{"okta.logs.read"},
+	}
+	got, err := generateOktaJWT([]byte(jwtText), cnf)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	tok, err := jwt.Parse([]byte(got), jwt.WithVerify(false))
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if tok.Issuer() != cnf.ClientID {
+		t.Errorf("unexpected issuer: got:%s want:%s", tok.Issuer(), cnf.ClientID)
+	}
+	if tok.Subject() != cnf.ClientID {
+		t.Errorf("unexpected subject: got:%s want:%s", tok.Subject(), cnf.ClientID)
+	}
+}
+
+func TestGenerateOktaJWTPEM(t *testing.T) {
+	// jwtText is generated by https://mkjwk.org/ using the instructions at
+	// https://developer.okta.com/docs/guides/dpop/nonoktaresourceserver/main/#create-the-json-web-token
+	const jwtText = `
+-----BEGIN PRIVATE KEY-----
+MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCOuef3HMRhohVT
+5kSoAJgV+atpDjkwTwkOq+ImnbBlv75GaApG90w8VpjXjhqN/1KJmwfyrKiquiMq
+OPu+o/672Dys5rUAaWSbT7wRF1GjLDDZrM0GHRdV4DGxM/LKI8I5yE1Mx3EzV+D5
+ZLmcRc5U4oEoMwtGpr0zRZ7uUr6a28UQwcUsVIPItc1/9rERlo1WTv8dcaj4ECC3
+2Sc0y/F+9XqwJvLd4Uv6ckzP0Sv4tbDA+7jpD9MneAIUiZ4LVj2cwbBd+YRY6jXx
+MkevcCSmSX60clBY1cIFkw1DYHqtdHEwAQcQHLGMoi72xRP2qrdzIPsaTKVYoHVo
+WA9vADdHAgMBAAECggEAIlx7jjCsztyYyeQsL05FTzUWoWo9NnYwtgmHnshkCXsK
+MiUmJEOxZO1sSqj5l6oakupyFWigCspZYPbrFNCiqVK7+NxqQzkccY/WtT6p9uDS
+ufUyPwCN96zMCd952lSVlBe3FH8Hr9a+YQxw60CbFjCZ67WuR0opTsi6JKJjJSDb
+TQQZ4qJR97D05I1TgfmO+VO7G/0/dDaNHnnlYz0AnOgZPSyvrU2G5cYye4842EMB
+ng81xjHD+xp55JNui/xYkhmYspYhrB2KlEjkKb08OInUjBeaLEAgA1r9yOHsfV/3
+DQzDPRO9iuqx5BfJhdIqUB1aifrye+sbxt9uMBtUgQKBgQDVdfO3GYT+ZycOQG9P
+QtdMn6uiSddchVCGFpk331u6M6yafCKjI/MlJDl29B+8R5sVsttwo8/qnV/xd3cn
+pY14HpKAsE4l6/Ciagzoj+0NqfPEDhEzbo8CyArcd7pSxt3XxECAfZe2+xivEPHe
+gFO60vSFjFtvlLRMDMOmqX3kYQKBgQCrK1DISyQTnD6/axsgh2/ESOmT7n+JRMx/
+YzA7Lxu3zGzUC8/sRDa1C41t054nf5ZXJueYLDSc4kEAPddzISuCLxFiTD2FQ75P
+lHWMgsEzQObDm4GPE9cdKOjoAvtAJwbvZcjDa029CDx7aCaDzbNvdmplZ7EUrznR
+55U8Wsm8pwKBgBytxTmzZwfbCgdDJvFKNKzpwuCB9TpL+v6Y6Kr2Clfg+26iAPFU
+MiWqUUInGGBuamqm5g6jI5sM28gQWeTsvC4IRXyes1Eq+uCHSQax15J/Y+3SSgNT
+9kjUYYkvWMwoRcPobRYWSZze7XkP2L8hFJ7EGvAaZGqAWxzgliS9HtnhAoGAONZ/
+UqMw7Zoac/Ga5mhSwrj7ZvXxP6Gqzjofj+eKqrOlB5yMhIX6LJATfH6iq7cAMxxm
+Fu/G4Ll4oB3o5wACtI3wldV/MDtYfJBtoCTjBqPsfNOsZ9hMvBATlsc2qwzKjsAb
+tFhzTevoOYpSD75EcSS/G8Ec2iN9bagatBnpl00CgYBVqAOFZelNfP7dj//lpk8y
+EUAw7ABOq0S9wkpFWTXIVPoBQUipm3iAUqGNPmvr/9ShdZC9xeu5AwKram4caMWJ
+ExRhcDP1hFM6CdmSkIYEgBKvN9N0O4Lx1ba34gk74Hm65KXxokjJHOC0plO7c7ok
+LNV/bIgMHOMoxiGrwyjAhg==
+-----END PRIVATE KEY-----
+`
+	cnf := &oauth2.Config{
+		ClientID: "0oaajljpeokFZLyKU5d7",
+		Scopes:   []string{"okta.logs.read"},
+	}
+	got, err := generateOktaJWTPEM(jwtText, cnf)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	tok, err := jwt.Parse([]byte(got), jwt.WithVerify(false))
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if tok.Issuer() != cnf.ClientID {
+		t.Errorf("unexpected issuer: got:%s want:%s", tok.Issuer(), cnf.ClientID)
+	}
+	if tok.Subject() != cnf.ClientID {
+		t.Errorf("unexpected subject: got:%s want:%s", tok.Subject(), cnf.ClientID)
+	}
+}

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

@@ -489,8 +489,8 @@ 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'"),
+		name:    "unique_okta_jwk_token",
+		wantErr: errors.New("okta validation error: one of okta.jwk_json, okta.jwk_file or okta.jwk_pem must be provided accessing 'auth.oauth2'"),
 		input: map[string]interface{}{
 			"auth.oauth2": map[string]interface{}{
 				"provider":  "okta",
@@ -501,7 +501,7 @@ var oAuth2ValidationTests = []struct {
 		},
 	},
 	{
-		name:    "okta oauth2 validation fails if jwk_json is not a valid JSON",
+		name:    "invalid_okta_jwk_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{}{
@@ -514,7 +514,7 @@ var oAuth2ValidationTests = []struct {
 		},
 	},
 	{
-		name: "okta successful oauth2 validation",
+		name: "okta_successful_oauth2_validation",
 		input: map[string]interface{}{
 			"auth.oauth2": map[string]interface{}{
 				"provider":      "okta",