x-pack/filebeat/input/{cel,httpjson,internal/dpop}: add support for DPoP OAuth for Okta

.github/CODEOWNERS

@@ -136,6 +136,7 @@ changelog/fragments/
 /x-pack/filebeat/input/gcs/ @elastic/security-service-integrations
 /x-pack/filebeat/input/http_endpoint/ @elastic/security-service-integrations
 /x-pack/filebeat/input/httpjson/ @elastic/security-service-integrations
+/x-pack/filebeat/input/internal/dpop @elastic/security-service-integrations
 /x-pack/filebeat/input/internal/httplog @elastic/security-service-integrations
 /x-pack/filebeat/input/internal/httpmon @elastic/security-service-integrations
 /x-pack/filebeat/input/internal/private @elastic/security-service-integrations

changelog/fragments/1762225281-15056-cel-httpjson.yaml

@@ -0,0 +1,17 @@
+kind: feature
+summary: Add support for DPoP authentication for the CEL and HTTP JSON inputs.
+component: filebeat
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+# pr: https://github.com/owner/repo/1234
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+# issue: https://github.com/owner/repo/1234

docs/reference/filebeat/filebeat-input-cel.md

@@ -715,6 +715,12 @@ Only one of the credentials settings can be set at once. For more information pl
 
 
 
+### `auth.oauth2.okta.dpop_key_pem` [_auth_oauth2_okta_dpop_key_pem]
+
+The Demonstrating Proof-of-Possession private key PEM block for your Okta authentication token. When this key is provided, Okta authentication will make use of the [Okta DPoP authentication flow](https://www.okta.com/blog/product-innovation/a-leap-forward-in-token-security-okta-adds-support-for-dpop/).
+
+
+
 ### `auth.token.enabled` [_auth_token_enabled]
 
 When set to `false`, disables the token authentication configuration. Default: `true`.

docs/reference/filebeat/filebeat-input-httpjson.md

@@ -407,6 +407,11 @@ The RSA JWK private key PEM block for your Okta Service App which is used for in
 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/](https://developer.okta.com/docs/guides/implement-oauth-for-okta-serviceapp/main/)
 ::::
 
+### `auth.oauth2.okta.dpop_key_pem` [_auth_oauth2_okta_dpop_key_pem_2]
+
+The Demonstrating Proof-of-Possession private key PEM block for your Okta authentication token. When this key is provided, Okta authentication will make use of the [Okta DPoP authentication flow](https://www.okta.com/blog/product-innovation/a-leap-forward-in-token-security-okta-adds-support-for-dpop/).
+
+
 
 
 ### `auth.oauth2.google.delegated_account` [_auth_oauth2_google_delegated_account_2]

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

@@ -175,6 +175,7 @@ type oAuth2Config struct {
 	OktaJWKFile string          `config:"okta.jwk_file"`
 	OktaJWKJSON common.JSONBlob `config:"okta.jwk_json"`
 	OktaJWKPEM  string          `config:"okta.jwk_pem"`
+	DPoPKeyPEM  string          `config:"okta.dpop_key_pem"`
 }
 
 // isEnabled returns true if the `enable` field is set to true in the yaml.
@@ -245,7 +246,7 @@ func (o *oAuth2Config) client(ctx context.Context, client *http.Client) (*http.C
 		}
 		return oauth2.NewClient(ctx, creds.TokenSource), nil
 	case oAuth2ProviderOkta:
-		return o.fetchOktaOauthClient(ctx, client)
+		return o.fetchOktaOauthClient(ctx)
 	default:
 		return nil, errors.New("oauth2 client: unknown provider")
 	}

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

@@ -7,6 +7,7 @@ package cel
 import (
 	"bytes"
 	"context"
+	"crypto"
 	"crypto/rsa"
 	"crypto/x509"
 	"encoding/base64"
@@ -24,6 +25,8 @@ import (
 	"github.com/golang-jwt/jwt/v5"
 	"golang.org/x/oauth2"
 	"golang.org/x/oauth2/clientcredentials"
+
+	"github.com/elastic/beats/v7/x-pack/filebeat/input/internal/dpop"
 )
 
 // oktaTokenSource is a custom implementation of the oauth2.TokenSource interface.
@@ -37,7 +40,7 @@ type oktaTokenSource struct {
 }
 
 // fetchOktaOauthClient fetches an OAuth2 client using the Okta JWK credentials.
-func (o *oAuth2Config) fetchOktaOauthClient(ctx context.Context, _ *http.Client) (*http.Client, error) {
+func (o *oAuth2Config) fetchOktaOauthClient(ctx context.Context) (*http.Client, error) {
 	conf := &oauth2.Config{
 		ClientID: o.ClientID,
 		Scopes:   o.Scopes,
@@ -46,11 +49,40 @@ func (o *oAuth2Config) fetchOktaOauthClient(ctx context.Context, _ *http.Client)
 		},
 	}
 
+	oauthCtx := ctx
+	var (
+		claim  dpop.ClaimerFunc
+		key    crypto.Signer
+		method jwt.SigningMethod
+	)
+	if o.DPoPKeyPEM != "" {
+		claim = func() *jwt.RegisteredClaims {
+			now := time.Now()
+			return &jwt.RegisteredClaims{
+				Audience:  []string{conf.Endpoint.TokenURL},
+				Issuer:    conf.ClientID,
+				Subject:   conf.ClientID,
+				IssuedAt:  jwt.NewNumericDate(now),
+				ExpiresAt: jwt.NewNumericDate(now.Add(time.Hour)),
+			}
+		}
+		var err error
+		key, err = pemPKCS8PrivateKey([]byte(o.DPoPKeyPEM))
+		if err != nil {
+			return nil, fmt.Errorf("failed to decode dpop signer: %w", err)
+		}
+		cli, err := dpop.NewTokenClient(claim, key, jwt.SigningMethodRS256, nil)
+		if err != nil {
+			return nil, fmt.Errorf("failed to make token client: %w", err)
+		}
+		oauthCtx = context.WithValue(oauthCtx, oauth2.HTTPClient, cli)
+	}
+
 	var (
 		oktaJWT string
 		err     error
 	)
-	if len(o.OktaJWKPEM) != 0 {
+	if o.OktaJWKPEM != "" {
 		oktaJWT, err = generateOktaJWTPEM(o.OktaJWKPEM, conf)
 		if err != nil {
 			return nil, fmt.Errorf("oauth2 client: error generating Okta JWT PEM: %w", err)
@@ -62,7 +94,7 @@ func (o *oAuth2Config) fetchOktaOauthClient(ctx context.Context, _ *http.Client)
 		}
 	}
 
-	token, err := exchangeForBearerToken(ctx, oktaJWT, conf)
+	token, err := exchangeForBearerToken(oauthCtx, oktaJWT, conf)
 	if err != nil {
 		return nil, fmt.Errorf("oauth2 client: error exchanging Okta JWT for bearer token: %w", err)
 	}
@@ -76,7 +108,9 @@ func (o *oAuth2Config) fetchOktaOauthClient(ctx context.Context, _ *http.Client)
 	// 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))
-
+	if claim != nil {
+		return dpop.NewResourceClient(claim, key, method, tokenSource, client)
+	}
 	return client, nil
 }
 
@@ -167,20 +201,28 @@ func generateOktaJWTPEM(pemdata string, cnf *oauth2.Config) (string, error) {
 	return signJWT(cnf, key)
 }
 
-func pemPKCS8PrivateKey(pemdata []byte) (any, error) {
+func pemPKCS8PrivateKey(pemdata []byte) (crypto.Signer, error) {
 	blk, rest := pem.Decode(pemdata)
 	if rest := bytes.TrimSpace(rest); len(rest) != 0 {
 		return nil, fmt.Errorf("PEM text has trailing data: %d bytes", len(rest))
 	}
 	if blk == nil {
 		return nil, errors.New("no PEM data")
 	}
-	return x509.ParsePKCS8PrivateKey(blk.Bytes)
+	key, err := x509.ParsePKCS8PrivateKey(blk.Bytes)
+	if err != nil {
+		return nil, err
+	}
+	signer, ok := key.(crypto.Signer)
+	if !ok {
+		return nil, fmt.Errorf("key is not a signer: %T", key)
+	}
+	return signer, nil
 }
 
 // signJWT creates a JWT token using required claims and sign it with the
 // private key.
-func signJWT(cnf *oauth2.Config, key any) (string, error) {
+func signJWT(cnf *oauth2.Config, key crypto.Signer) (string, error) {
 	now := time.Now()
 	signed, err := jwt.NewWithClaims(jwt.SigningMethodRS256, jwt.RegisteredClaims{
 		Audience:  []string{cnf.Endpoint.TokenURL},

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

@@ -117,6 +117,7 @@ type oAuth2Config struct {
 	OktaJWKFile string          `config:"okta.jwk_file"`
 	OktaJWKJSON common.JSONBlob `config:"okta.jwk_json"`
 	OktaJWKPEM  string          `config:"okta.jwk_pem"`
+	DPoPKeyPEM  string          `config:"okta.dpop_key_pem"`
 
 	prepared *http.Client
 }
@@ -191,7 +192,7 @@ func (o *oAuth2Config) client(ctx context.Context, client *http.Client) (*http.C
 		}
 		return oauth2.NewClient(ctx, creds.TokenSource), nil
 	case oAuth2ProviderOkta:
-		return o.fetchOktaOauthClient(ctx, client)
+		return o.fetchOktaOauthClient(ctx)
 
 	default:
 		return nil, errors.New("oauth2 client: unknown provider")

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

@@ -7,6 +7,7 @@ package httpjson
 import (
 	"bytes"
 	"context"
+	"crypto"
 	"crypto/rsa"
 	"crypto/x509"
 	"encoding/base64"
@@ -24,6 +25,8 @@ import (
 	"github.com/golang-jwt/jwt/v5"
 	"golang.org/x/oauth2"
 	"golang.org/x/oauth2/clientcredentials"
+
+	"github.com/elastic/beats/v7/x-pack/filebeat/input/internal/dpop"
 )
 
 // oktaTokenSource is a custom implementation of the oauth2.TokenSource interface.
@@ -37,7 +40,7 @@ type oktaTokenSource struct {
 }
 
 // fetchOktaOauthClient fetches an OAuth2 client using the Okta JWK credentials.
-func (o *oAuth2Config) fetchOktaOauthClient(ctx context.Context, _ *http.Client) (*http.Client, error) {
+func (o *oAuth2Config) fetchOktaOauthClient(ctx context.Context) (*http.Client, error) {
 	conf := &oauth2.Config{
 		ClientID: o.ClientID,
 		Scopes:   o.Scopes,
@@ -46,11 +49,40 @@ func (o *oAuth2Config) fetchOktaOauthClient(ctx context.Context, _ *http.Client)
 		},
 	}
 
+	oauthCtx := ctx
+	var (
+		claim  dpop.ClaimerFunc
+		key    crypto.Signer
+		method jwt.SigningMethod
+	)
+	if o.DPoPKeyPEM != "" {
+		claim = func() *jwt.RegisteredClaims {
+			now := time.Now()
+			return &jwt.RegisteredClaims{
+				Audience:  []string{conf.Endpoint.TokenURL},
+				Issuer:    conf.ClientID,
+				Subject:   conf.ClientID,
+				IssuedAt:  jwt.NewNumericDate(now),
+				ExpiresAt: jwt.NewNumericDate(now.Add(time.Hour)),
+			}
+		}
+		var err error
+		key, err = pemPKCS8PrivateKey([]byte(o.DPoPKeyPEM))
+		if err != nil {
+			return nil, fmt.Errorf("failed to decode dpop signer: %w", err)
+		}
+		cli, err := dpop.NewTokenClient(claim, key, jwt.SigningMethodRS256, nil)
+		if err != nil {
+			return nil, fmt.Errorf("failed to make token client: %w", err)
+		}
+		oauthCtx = context.WithValue(oauthCtx, oauth2.HTTPClient, cli)
+	}
+
 	var (
 		oktaJWT string
 		err     error
 	)
-	if len(o.OktaJWKPEM) != 0 {
+	if o.OktaJWKPEM != "" {
 		oktaJWT, err = generateOktaJWTPEM(o.OktaJWKPEM, conf)
 		if err != nil {
 			return nil, fmt.Errorf("oauth2 client: error generating Okta JWT PEM: %w", err)
@@ -62,7 +94,7 @@ func (o *oAuth2Config) fetchOktaOauthClient(ctx context.Context, _ *http.Client)
 		}
 	}
 
-	token, err := exchangeForBearerToken(ctx, oktaJWT, conf)
+	token, err := exchangeForBearerToken(oauthCtx, oktaJWT, conf)
 	if err != nil {
 		return nil, fmt.Errorf("oauth2 client: error exchanging Okta JWT for bearer token: %w", err)
 	}
@@ -75,7 +107,9 @@ func (o *oAuth2Config) fetchOktaOauthClient(ctx context.Context, _ *http.Client)
 	}
 	// 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))
-
+	if claim != nil {
+		return dpop.NewResourceClient(claim, key, method, tokenSource, client)
+	}
 	return client, nil
 }
 
@@ -165,15 +199,23 @@ func generateOktaJWTPEM(pemdata string, cnf *oauth2.Config) (string, error) {
 	return signJWT(cnf, key)
 }
 
-func pemPKCS8PrivateKey(pemdata []byte) (any, error) {
+func pemPKCS8PrivateKey(pemdata []byte) (crypto.Signer, error) {
 	blk, rest := pem.Decode(pemdata)
 	if rest := bytes.TrimSpace(rest); len(rest) != 0 {
 		return nil, fmt.Errorf("PEM text has trailing data: %d bytes", len(rest))
 	}
 	if blk == nil {
 		return nil, errors.New("no PEM data")
 	}
-	return x509.ParsePKCS8PrivateKey(blk.Bytes)
+	key, err := x509.ParsePKCS8PrivateKey(blk.Bytes)
+	if err != nil {
+		return nil, err
+	}
+	signer, ok := key.(crypto.Signer)
+	if !ok {
+		return nil, fmt.Errorf("key is not a signer: %T", key)
+	}
+	return signer, nil
 }
 
 // signJWT creates a JWT token using required claims and sign it with the private key.