feat(action v2): add JWT and JWE payload type options (#11196)

# Which Problems Are Solved

The payload in actions V2 is currently sent as JSON to the target
endpoint. It might get exposed to intermediary infrastructure or logging
systems.
For these scenarios there needs to be an application-layer encryption,
where the provider of the endpoint can define an key to be used for the
encryption.

# How the Problems Are Solved

- Added an additional option to the target to specify the payload type:
`JSON` (current and default), `JWT`, `JWE` (api and console)
- added endpoints to upload and manage public keys (to be used for
encryption) for a target
- updated all action v2 executions (interceptors, oidc, saml, ...) to
provide the `GetActiveSigningWebKey` from queries
- implemented jwt and jwe in the exections incl. refactoring of code and
tests
- changes to the authn_keys table:
  - added a `fingerprint` column
  - dropped not null constraint on expiration
- moved the `GetSignerOnce` into its own package to prevent circular
dependencies

# Additional Changes

None

# Additional Context

closes #11061

---------

Co-authored-by: Marco A. <[email protected]>
Co-authored-by: conblem <[email protected]>
Co-authored-by: Silvan <[email protected]>

internal/execution/execution.go

@@ -3,15 +3,22 @@ package execution
 import (
 	"bytes"
 	"context"
+	"crypto/ecdsa"
+	"crypto/rsa"
+	"crypto/x509"
 	"encoding/json"
+	"encoding/pem"
 	"io"
 	"net/http"
+	"sync"
 	"time"
 
+	"github.com/go-jose/go-jose/v4"
 	"github.com/zitadel/logging"
 
 	"github.com/zitadel/zitadel/internal/api/authz"
 	zhttp "github.com/zitadel/zitadel/internal/api/http"
+	"github.com/zitadel/zitadel/internal/api/oidc/sign"
 	"github.com/zitadel/zitadel/internal/crypto"
 	"github.com/zitadel/zitadel/internal/domain"
 	target_domain "github.com/zitadel/zitadel/internal/execution/target"
@@ -27,20 +34,29 @@ type ContextInfo interface {
 	SetHTTPResponseBody([]byte) error
 }
 
+type GetActiveSigningWebKey func(ctx context.Context) (*jose.JSONWebKey, error)
+
 // CallTargets call a list of targets in order with handling of error and responses
 func CallTargets(
 	ctx context.Context,
 	targets []target_domain.Target,
 	info ContextInfo,
 	alg crypto.EncryptionAlgorithm,
+	activeSigningKey GetActiveSigningWebKey,
 ) (_ interface{}, err error) {
 	ctx, span := tracing.NewSpan(ctx)
 	defer func() { span.EndWithError(err) }()
 
+	// We make sure the signer and its key are only fetched once per CallTargets call.
+	signerOnce := sign.GetSignerOnce(activeSigningKey)
+	// Create a map to cache encrypters by key ID to avoid recreating them for each target.
+	encrypters := &sync.Map{}
+
 	for _, target := range targets {
 		// call the type of target
-		resp, err := CallTarget(ctx, target, info, alg)
+		resp, err := CallTarget(ctx, target, info, alg, signerOnce, encrypters)
 		// handle error if interrupt is set
+		logging.WithFields("instanceID", authz.GetInstance(ctx).InstanceID(), "target", target.GetTargetID()).OnError(err).Error("error calling target")
 		if err != nil && target.IsInterruptOnError() {
 			return nil, err
 		}
@@ -64,6 +80,8 @@ func CallTarget(
 	target target_domain.Target,
 	info ContextInfoRequest,
 	alg crypto.EncryptionAlgorithm,
+	signerOnce sign.SignerFunc,
+	encrypters *sync.Map,
 ) (res []byte, err error) {
 	ctx, span := tracing.NewSpan(ctx)
 	defer func() { span.EndWithError(err) }()
@@ -72,26 +90,137 @@ func CallTarget(
 	if err != nil {
 		return nil, zerrors.ThrowInternal(err, "EXEC-thiiCh5b", "Errors.Internal")
 	}
+	body, err := payload(ctx, info.GetHTTPRequestBody(), target, signerOnce, encrypters)
+	if err != nil {
+		return nil, err
+	}
 
 	switch target.GetTargetType() {
 	// get request, ignore response and return request and error for handling in list of targets
 	case target_domain.TargetTypeWebhook:
-		return nil, webhook(ctx, target.GetEndpoint(), target.GetTimeout(), info.GetHTTPRequestBody(), signingKey)
+		return nil, webhook(ctx, target.GetEndpoint(), target.GetTimeout(), body, signingKey)
 	// get request, return response and error
 	case target_domain.TargetTypeCall:
-		return Call(ctx, target.GetEndpoint(), target.GetTimeout(), info.GetHTTPRequestBody(), signingKey)
+		return Call(ctx, target.GetEndpoint(), target.GetTimeout(), body, signingKey)
 	case target_domain.TargetTypeAsync:
 		go func(ctx context.Context, target target_domain.Target, info []byte) {
 			if _, err := Call(ctx, target.GetEndpoint(), target.GetTimeout(), info, signingKey); err != nil {
 				logging.WithFields("target", target.GetTargetID()).OnError(err).Info(err)
 			}
-		}(context.WithoutCancel(ctx), target, info.GetHTTPRequestBody())
+		}(context.WithoutCancel(ctx), target, body)
 		return nil, nil
 	default:
 		return nil, zerrors.ThrowInternal(nil, "EXEC-auqnansr2m", "Errors.Execution.Unknown")
 	}
 }
 
+func payload(ctx context.Context, payload []byte, target target_domain.Target, signerOnce sign.SignerFunc, encrypters *sync.Map) ([]byte, error) {
+	switch target.GetPayloadType() {
+	case target_domain.PayloadTypeUnspecified,
+		target_domain.PayloadTypeJSON:
+		return payload, nil
+	case target_domain.PayloadTypeJWT:
+		return payloadJWT(ctx, payload, signerOnce)
+	case target_domain.PayloadTypeJWE:
+		return payloadJWE(ctx, payload, target, signerOnce, encrypters)
+	default:
+		return payload, nil
+	}
+}
+
+func payloadJWT(ctx context.Context, payload []byte, signerOnce sign.SignerFunc) ([]byte, error) {
+	signer, _, err := signerOnce(ctx)
+	if err != nil {
+		return nil, err
+	}
+	sig, err := signer.Sign(payload)
+	if err != nil {
+		return nil, err
+	}
+	data, err := sig.CompactSerialize()
+	if err != nil {
+		return nil, err
+	}
+	return []byte(data), nil
+}
+
+func loadEncrypter(target target_domain.Target, encrypters *sync.Map) (jose.Encrypter, error) {
+	if encrypter, ok := encrypters.Load(target.GetEncryptionKeyID()); ok {
+		return encrypter.(jose.Encrypter), nil
+	}
+	encryptionKey := target.GetEncryptionKey()
+	if len(encryptionKey) == 0 {
+		return nil, zerrors.ThrowInternal(nil, "EXEC-2n8fhs7g", "Errors.Execution.MissingEncryptionKey")
+	}
+	publicKey, algorithm, err := publicKeyFromBytes(encryptionKey)
+	if err != nil {
+		return nil, err
+	}
+
+	encrypter, err := jose.NewEncrypter(
+		jose.A256GCM,
+		jose.Recipient{
+			Algorithm: algorithm,
+			Key:       publicKey,
+			KeyID:     target.GetEncryptionKeyID(),
+		},
+		(&jose.EncrypterOptions{}).
+			WithType("JWT").
+			WithContentType("JWT"),
+	)
+	if err != nil {
+		return nil, err
+	}
+	encrypters.Store(target.GetEncryptionKeyID(), encrypter)
+	return encrypter, nil
+}
+
+func payloadJWE(
+	ctx context.Context,
+	payload []byte,
+	target target_domain.Target,
+	signerOnce sign.SignerFunc,
+	encrypters *sync.Map,
+) ([]byte, error) {
+	payload, err := payloadJWT(ctx, payload, signerOnce)
+	if err != nil {
+		return nil, err
+	}
+	encrypter, err := loadEncrypter(target, encrypters)
+	if err != nil {
+		return nil, err
+	}
+
+	enc, err := encrypter.Encrypt(payload)
+	if err != nil {
+		return nil, err
+	}
+	crypted, err := enc.CompactSerialize()
+	if err != nil {
+		return nil, err
+	}
+	return []byte(crypted), nil
+}
+
+func publicKeyFromBytes(data []byte) (any, jose.KeyAlgorithm, error) {
+	block, _ := pem.Decode(data)
+	if block == nil {
+		return nil, "", zerrors.ThrowInternal(nil, "EXEC-3n8fhs7g", "Errors.Execution.InvalidPublicKey")
+	}
+	publicKey, err := x509.ParsePKIXPublicKey(block.Bytes)
+	if err != nil {
+		return nil, "", err
+	}
+	switch pk := publicKey.(type) {
+	case *rsa.PublicKey:
+		return pk, jose.RSA_OAEP_256, nil
+	case *ecdsa.PublicKey:
+		return pk, jose.ECDH_ES_A256KW, nil
+	default:
+		return nil, "", zerrors.ThrowInternal(nil, "EXEC-NKJe2", "Errors.Execution.InvalidPublicKey")
+	}
+}
+
 // webhook call a webhook, ignore the response but return the errror
 func webhook(ctx context.Context, url string, timeout time.Duration, body []byte, signingKey string) error {
 	_, err := Call(ctx, url, timeout, body, signingKey)

internal/execution/execution_test.go

@@ -3,20 +3,26 @@ package execution_test
 import (
 	"bytes"
 	"context"
+	"crypto/rand"
+	"crypto/rsa"
 	"encoding/json"
 	"errors"
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
+	"github.com/go-jose/go-jose/v4"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/mock/gomock"
 	"google.golang.org/protobuf/types/known/structpb"
 
 	"github.com/zitadel/zitadel/internal/api/grpc/server/middleware"
+	"github.com/zitadel/zitadel/internal/api/oidc/sign"
 	"github.com/zitadel/zitadel/internal/crypto"
 	"github.com/zitadel/zitadel/internal/execution"
 	target_domain "github.com/zitadel/zitadel/internal/execution/target"
@@ -111,7 +117,7 @@ func Test_Call(t *testing.T) {
 			respBody, err := testServer(t,
 				&callTestServer{
 					method:      tt.args.method,
-					expectBody:  tt.args.body,
+					expectBody:  validateJSONPayload(tt.args.body),
 					timeout:     tt.args.sleep,
 					statusCode:  tt.args.statusCode,
 					respondBody: tt.args.respBody,
@@ -135,6 +141,7 @@ func Test_CallTarget(t *testing.T) {
 		info   *middleware.ContextInfoRequest
 		server *callTestServer
 		target target_domain.Target
+		signer func(ctx context.Context) (jose.Signer, jose.SignatureAlgorithm, error)
 	}
 	type res struct {
 		body    []byte
@@ -152,7 +159,7 @@ func Test_CallTarget(t *testing.T) {
 				info: requestContextInfo1,
 				server: &callTestServer{
 					method:      http.MethodPost,
-					expectBody:  []byte("{\"request\":{\"content\":\"request1\"}}"),
+					expectBody:  validateJSONPayload([]byte("{\"request\":{\"content\":\"request1\"}}")),
 					respondBody: []byte("{\"content\":\"request2\"}"),
 					timeout:     time.Second,
 					statusCode:  http.StatusInternalServerError,
@@ -173,7 +180,7 @@ func Test_CallTarget(t *testing.T) {
 				server: &callTestServer{
 					timeout:     time.Second,
 					method:      http.MethodPost,
-					expectBody:  []byte("{\"request\":{\"content\":\"request1\"}}"),
+					expectBody:  validateJSONPayload([]byte("{\"request\":{\"content\":\"request1\"}}")),
 					respondBody: []byte("{\"content\":\"request2\"}"),
 					statusCode:  http.StatusInternalServerError,
 				},
@@ -194,7 +201,7 @@ func Test_CallTarget(t *testing.T) {
 				server: &callTestServer{
 					timeout:     time.Second,
 					method:      http.MethodPost,
-					expectBody:  []byte("{\"request\":{\"content\":\"request1\"}}"),
+					expectBody:  validateJSONPayload([]byte("{\"request\":{\"content\":\"request1\"}}")),
 					respondBody: []byte("{\"content\":\"request2\"}"),
 					statusCode:  http.StatusOK,
 				},
@@ -215,7 +222,7 @@ func Test_CallTarget(t *testing.T) {
 				server: &callTestServer{
 					timeout:     time.Second,
 					method:      http.MethodPost,
-					expectBody:  []byte("{\"request\":{\"content\":\"request1\"}}"),
+					expectBody:  validateJSONPayload([]byte("{\"request\":{\"content\":\"request1\"}}")),
 					respondBody: []byte("{\"content\":\"request2\"}"),
 					statusCode:  http.StatusOK,
 					signingKey:  "signingkey",
@@ -242,7 +249,7 @@ func Test_CallTarget(t *testing.T) {
 				server: &callTestServer{
 					timeout:     time.Second,
 					method:      http.MethodPost,
-					expectBody:  []byte("{\"request\":{\"content\":\"request1\"}}"),
+					expectBody:  validateJSONPayload([]byte("{\"request\":{\"content\":\"request1\"}}")),
 					respondBody: []byte("{\"content\":\"request2\"}"),
 					statusCode:  http.StatusInternalServerError,
 				},
@@ -263,7 +270,7 @@ func Test_CallTarget(t *testing.T) {
 				server: &callTestServer{
 					timeout:     time.Second,
 					method:      http.MethodPost,
-					expectBody:  []byte("{\"request\":{\"content\":\"request1\"}}"),
+					expectBody:  validateJSONPayload([]byte("{\"request\":{\"content\":\"request1\"}}")),
 					respondBody: []byte("{\"content\":\"request2\"}"),
 					statusCode:  http.StatusOK,
 				},
@@ -284,7 +291,7 @@ func Test_CallTarget(t *testing.T) {
 				server: &callTestServer{
 					timeout:     time.Second,
 					method:      http.MethodPost,
-					expectBody:  []byte("{\"request\":{\"content\":\"request1\"}}"),
+					expectBody:  validateJSONPayload([]byte("{\"request\":{\"content\":\"request1\"}}")),
 					respondBody: []byte("{\"content\":\"request2\"}"),
 					statusCode:  http.StatusOK,
 					signingKey:  "signingkey",
@@ -303,10 +310,103 @@ func Test_CallTarget(t *testing.T) {
 				body: []byte("{\"content\":\"request2\"}"),
 			},
 		},
+		{
+			"webhook, JWT, ok",
+			args{
+				ctx:  context.Background(),
+				info: requestContextInfo1,
+				server: &callTestServer{
+					timeout:     time.Second,
+					method:      http.MethodPost,
+					expectBody:  validateJWTPayload([]byte("{\"request\":{\"content\":\"request1\"}}")),
+					respondBody: []byte("{\"content\":\"request2\"}"),
+					statusCode:  http.StatusOK,
+					signingKey:  "signingkey",
+				},
+				target: target_domain.Target{
+					TargetType: target_domain.TargetTypeWebhook,
+					Timeout:    time.Minute,
+					SigningKey: &crypto.CryptoValue{
+						Algorithm: "enc",
+						KeyID:     "id",
+						Crypted:   []byte("signingkey"),
+					},
+					PayloadType: target_domain.PayloadTypeJWT,
+				},
+				signer: mockSigner,
+			},
+			res{
+				body: nil,
+			},
+		},
+		{
+			"webhook, JWE, ok",
+			args{
+				ctx:  context.Background(),
+				info: requestContextInfo1,
+				server: &callTestServer{
+					timeout:     time.Second,
+					method:      http.MethodPost,
+					expectBody:  validateJWEPayload([]byte("{\"request\":{\"content\":\"request1\"}}")),
+					respondBody: []byte("{\"content\":\"request2\"}"),
+					statusCode:  http.StatusOK,
+					signingKey:  "signingkey",
+				},
+				target: target_domain.Target{
+					TargetType: target_domain.TargetTypeWebhook,
+					Timeout:    time.Minute,
+					SigningKey: &crypto.CryptoValue{
+						Algorithm: "enc",
+						KeyID:     "id",
+						Crypted:   []byte("signingkey"),
+					},
+					PayloadType:     target_domain.PayloadTypeJWE,
+					EncryptionKey:   encryptionKey,
+					EncryptionKeyID: encryptionKeyID,
+				},
+				signer: mockSigner,
+			},
+			res{
+				body: nil,
+			},
+		},
+		{
+			"webhook, JWE no encryption key, error",
+			args{
+				ctx:  context.Background(),
+				info: requestContextInfo1,
+				server: &callTestServer{
+					timeout:     time.Second,
+					method:      http.MethodPost,
+					expectBody:  validateJWEPayload([]byte("{\"request\":{\"content\":\"request1\"}}")),
+					respondBody: []byte("{\"content\":\"request2\"}"),
+					statusCode:  http.StatusOK,
+					signingKey:  "signingkey",
+				},
+				target: target_domain.Target{
+					TargetType: target_domain.TargetTypeWebhook,
+					Timeout:    time.Minute,
+					SigningKey: &crypto.CryptoValue{
+						Algorithm: "enc",
+						KeyID:     "id",
+						Crypted:   []byte("signingkey"),
+					},
+					PayloadType: target_domain.PayloadTypeJWE,
+				},
+				signer: mockSigner,
+			},
+			res{
+				wantErr: true,
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			respBody, err := testServer(t, tt.args.server, testCallTarget(tt.args.ctx, tt.args.info, tt.args.target, crypto.CreateMockEncryptionAlg(gomock.NewController(t))))
+			respBody, err := testServer(
+				t,
+				tt.args.server,
+				testCallTarget(tt.args.ctx, tt.args.info, tt.args.target, crypto.CreateMockEncryptionAlg(gomock.NewController(t)), tt.args.signer, &sync.Map{}),
+			)
 			if tt.res.wantErr {
 				assert.Error(t, err)
 			} else {
@@ -319,10 +419,11 @@ func Test_CallTarget(t *testing.T) {
 
 func Test_CallTargets(t *testing.T) {
 	type args struct {
-		ctx     context.Context
-		info    *middleware.ContextInfoRequest
-		servers []*callTestServer
-		targets []target_domain.Target
+		ctx                    context.Context
+		info                   *middleware.ContextInfoRequest
+		servers                []*callTestServer
+		targets                []target_domain.Target
+		getActiveSigningWebKey func(*int32) execution.GetActiveSigningWebKey
 	}
 	type res struct {
 		ret     interface{}
@@ -341,13 +442,13 @@ func Test_CallTargets(t *testing.T) {
 				servers: []*callTestServer{{
 					timeout:     time.Second,
 					method:      http.MethodPost,
-					expectBody:  requestContextInfoBody1,
+					expectBody:  validateJSONPayload(requestContextInfoBody1),
 					respondBody: requestContextInfoBody2,
 					statusCode:  http.StatusInternalServerError,
 				}, {
 					timeout:     time.Second,
 					method:      http.MethodPost,
-					expectBody:  requestContextInfoBody1,
+					expectBody:  validateJSONPayload(requestContextInfoBody1),
 					respondBody: requestContextInfoBody2,
 					statusCode:  http.StatusInternalServerError,
 				}},
@@ -368,13 +469,13 @@ func Test_CallTargets(t *testing.T) {
 				servers: []*callTestServer{{
 					timeout:     time.Second,
 					method:      http.MethodPost,
-					expectBody:  requestContextInfoBody1,
+					expectBody:  validateJSONPayload(requestContextInfoBody1),
 					respondBody: requestContextInfoBody2,
 					statusCode:  http.StatusInternalServerError,
 				}, {
 					timeout:     time.Second,
 					method:      http.MethodPost,
-					expectBody:  requestContextInfoBody1,
+					expectBody:  validateJSONPayload(requestContextInfoBody1),
 					respondBody: requestContextInfoBody2,
 					statusCode:  http.StatusInternalServerError,
 				}},
@@ -395,13 +496,13 @@ func Test_CallTargets(t *testing.T) {
 				servers: []*callTestServer{{
 					timeout:     time.Second,
 					method:      http.MethodPost,
-					expectBody:  requestContextInfoBody1,
+					expectBody:  validateJSONPayload(requestContextInfoBody1),
 					respondBody: requestContextInfoBody2,
 					statusCode:  http.StatusOK,
 				}, {
 					timeout:     time.Second,
 					method:      http.MethodPost,
-					expectBody:  requestContextInfoBody1,
+					expectBody:  validateJSONPayload(requestContextInfoBody1),
 					respondBody: []byte("just a string, not json"),
 					statusCode:  http.StatusOK,
 				}},
@@ -422,13 +523,13 @@ func Test_CallTargets(t *testing.T) {
 				servers: []*callTestServer{{
 					timeout:     time.Second,
 					method:      http.MethodPost,
-					expectBody:  requestContextInfoBody1,
+					expectBody:  validateJSONPayload(requestContextInfoBody1),
 					respondBody: requestContextInfoBody2,
 					statusCode:  http.StatusOK,
 				}, {
 					timeout:     time.Second,
 					method:      http.MethodPost,
-					expectBody:  requestContextInfoBody1,
+					expectBody:  validateJSONPayload(requestContextInfoBody1),
 					respondBody: []byte("just a string, not json"),
 					statusCode:  http.StatusOK,
 				}},
@@ -440,12 +541,53 @@ func Test_CallTargets(t *testing.T) {
 				ret: requestContextInfo1.GetContent(),
 			},
 		},
+		{
+			"multiple JWT/JWE targets, ok",
+			args{
+				ctx:  context.Background(),
+				info: requestContextInfo1,
+				servers: []*callTestServer{{
+					timeout:     time.Second,
+					method:      http.MethodPost,
+					expectBody:  validateJWTPayload(requestContextInfoBody1),
+					respondBody: requestContextInfoBody2,
+					statusCode:  http.StatusOK,
+				}, {
+					timeout:     time.Second,
+					method:      http.MethodPost,
+					expectBody:  validateJSONPayload(requestContextInfoBody1),
+					respondBody: requestContextInfoBody2,
+					statusCode:  http.StatusOK,
+				}},
+				targets: []target_domain.Target{
+					{
+						TargetType:  target_domain.TargetTypeWebhook,
+						PayloadType: target_domain.PayloadTypeJWT,
+						Timeout:     time.Minute,
+					},
+					{
+						TargetType:  target_domain.TargetTypeWebhook,
+						PayloadType: target_domain.PayloadTypeJWE,
+						Timeout:     time.Minute,
+					},
+				},
+				getActiveSigningWebKey: testActiveSingingWebKey,
+			},
+			res{
+				ret: requestContextInfo1.GetContent(),
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			var getWebKeyCalls int32
+			var getActiveSigningWebKey execution.GetActiveSigningWebKey
+			if tt.args.getActiveSigningWebKey != nil {
+				getActiveSigningWebKey = tt.args.getActiveSigningWebKey(&getWebKeyCalls)
+			}
 			respBody, err := testServers(t,
 				tt.args.servers,
-				testCallTargets(tt.args.ctx, tt.args.info, tt.args.targets, crypto.CreateMockEncryptionAlg(gomock.NewController(t))),
+				testCallTargets(tt.args.ctx, tt.args.info, tt.args.targets, crypto.CreateMockEncryptionAlg(gomock.NewController(t)), getActiveSigningWebKey),
 			)
 			if tt.res.wantErr {
 				assert.Error(t, err)
@@ -453,17 +595,27 @@ func Test_CallTargets(t *testing.T) {
 				assert.NoError(t, err)
 			}
 			assert.Equal(t, tt.res.ret, respBody)
+			var expectedCalls int32
+			if tt.args.getActiveSigningWebKey != nil {
+				expectedCalls = 1
+			}
+			assert.Equal(t, expectedCalls, atomic.LoadInt32(&getWebKeyCalls))
 		})
 	}
 }
 
 type callTestServer struct {
 	method      string
-	expectBody  []byte
+	expectBody  func(*testing.T, []byte) bool
 	timeout     time.Duration
 	statusCode  int
 	respondBody []byte
 	signingKey  string
+	called      bool
+}
+
+func (s *callTestServer) Called() bool {
+	return s.called
 }
 
 func testServers(
@@ -473,8 +625,8 @@ func testServers(
 ) (interface{}, error) {
 	urls := make([]string, len(c))
 	for i := range c {
-		url, close := listen(t, c[i])
-		defer close()
+		url, closeF, _ := listen(t, c[i])
+		defer closeF()
 		urls[i] = url
 	}
 	return call(urls)
@@ -485,16 +637,17 @@ func testServer(
 	c *callTestServer,
 	call func(string) ([]byte, error),
 ) ([]byte, error) {
-	url, close := listen(t, c)
-	defer close()
+	url, closeF, _ := listen(t, c)
+	defer closeF()
 	return call(url)
 }
 
 func listen(
 	t *testing.T,
 	c *callTestServer,
-) (url string, close func()) {
+) (url string, close func(), called func() bool) {
 	handler := func(w http.ResponseWriter, r *http.Request) {
+		c.called = true
 		checkRequest(t, r, c.method, c.expectBody, c.signingKey)
 
 		if c.statusCode != http.StatusOK {
@@ -511,13 +664,15 @@ func listen(
 		}
 	}
 	server := httptest.NewServer(http.HandlerFunc(handler))
-	return server.URL, server.Close
+	return server.URL, server.Close, c.Called
 }
 
-func checkRequest(t *testing.T, sent *http.Request, method string, expectedBody []byte, signingKey string) {
+func checkRequest(t *testing.T, sent *http.Request, method string, checkExpectedBody func(*testing.T, []byte) bool, signingKey string) {
 	sentBody, err := io.ReadAll(sent.Body)
 	require.NoError(t, err)
-	require.Equal(t, expectedBody, sentBody)
+	if !checkExpectedBody(t, sentBody) {
+		return
+	}
 	require.Equal(t, method, sent.Method)
 	if signingKey != "" {
 		require.NoError(t, actions.ValidatePayload(sentBody, sent.Header.Get(actions.SigningHeader), signingKey))
@@ -534,17 +689,20 @@ func testCallTarget(ctx context.Context,
 	info *middleware.ContextInfoRequest,
 	target target_domain.Target,
 	alg crypto.EncryptionAlgorithm,
+	signerOnce sign.SignerFunc,
+	encrypters *sync.Map,
 ) func(string) ([]byte, error) {
 	return func(url string) (r []byte, err error) {
 		target.Endpoint = url
-		return execution.CallTarget(ctx, target, info, alg)
+		return execution.CallTarget(ctx, target, info, alg, signerOnce, encrypters)
 	}
 }
 
 func testCallTargets(ctx context.Context,
-	info *middleware.ContextInfoRequest,
+	info execution.ContextInfo,
 	target []target_domain.Target,
 	alg crypto.EncryptionAlgorithm,
+	activeSigningKey execution.GetActiveSigningWebKey,
 ) func([]string) (interface{}, error) {
 	return func(urls []string) (interface{}, error) {
 		targets := make([]target_domain.Target, len(target))
@@ -552,7 +710,7 @@ func testCallTargets(ctx context.Context,
 			t.Endpoint = urls[i]
 			targets[i] = t
 		}
-		return execution.CallTargets(ctx, targets, info, alg)
+		return execution.CallTargets(ctx, targets, info, alg, activeSigningKey)
 	}
 }
 
@@ -717,3 +875,67 @@ func Test_handleResponse(t *testing.T) {
 	}
 
 }
+
+var (
+	privateKey = func() *rsa.PrivateKey {
+		privateKey, _ := rsa.GenerateKey(rand.Reader, 2048)
+		return privateKey
+	}()
+	encryptionKey = func() []byte {
+		data, _ := crypto.PublicKeyToBytes(&privateKey.PublicKey)
+		return data
+	}()
+	encryptionKeyID  = "encryption-key-id"
+	signingAlgorithm = jose.RS256
+)
+
+func mockSigner(ctx context.Context) (jose.Signer, jose.SignatureAlgorithm, error) {
+	return sign.GetSignerOnce(mockGetActiveSigningWebKey)(ctx)
+}
+
+func mockGetActiveSigningWebKey(ctx context.Context) (*jose.JSONWebKey, error) {
+	return &jose.JSONWebKey{
+		Key:       privateKey,
+		Algorithm: string(signingAlgorithm),
+		Use:       "sig",
+	}, nil
+}
+
+func testActiveSingingWebKey(getWebKeyCalls *int32) execution.GetActiveSigningWebKey {
+	return func(ctx context.Context) (*jose.JSONWebKey, error) {
+		atomic.AddInt32(getWebKeyCalls, 1)
+		return mockGetActiveSigningWebKey(ctx)
+	}
+}
+
+func validateJSONPayload(expected []byte) func(*testing.T, []byte) bool {
+	return func(t *testing.T, actual []byte) bool {
+		require.Equal(t, expected, actual)
+		return false
+	}
+}
+
+func validateJWTPayload(expected []byte) func(*testing.T, []byte) bool {
+	return func(t *testing.T, actual []byte) bool {
+		jws, err := jose.ParseSigned(string(actual), []jose.SignatureAlgorithm{jose.RS256})
+		require.NoError(t, err)
+		payload, err := jws.Verify(privateKey.Public())
+		require.NoError(t, err)
+		return bytes.Equal(expected, payload)
+	}
+}
+
+func validateJWEPayload(expected []byte) func(*testing.T, []byte) bool {
+	return func(t *testing.T, actual []byte) bool {
+		parsedJWE, err := jose.ParseEncrypted(string(actual), []jose.KeyAlgorithm{jose.RSA_OAEP_256, jose.ECDH_ES_A256KW}, []jose.ContentEncryption{jose.A256GCM})
+		if err != nil {
+			return false
+		}
+		require.Equal(t, encryptionKeyID, parsedJWE.Header.KeyID)
+		require.Equal(t, "JWT", parsedJWE.Header.ExtraHeaders[jose.HeaderContentType].(string))
+
+		decryptedJWS, err := parsedJWE.Decrypt(privateKey)
+		require.NoError(t, err)
+		return validateJWTPayload(expected)(t, decryptedJWS)
+	}
+}

internal/execution/projections.go

@@ -2,6 +2,7 @@ package execution
 
 import (
 	"context"
+	"time"
 
 	"github.com/zitadel/zitadel/internal/crypto"
 	"github.com/zitadel/zitadel/internal/eventstore/handler/v2"
@@ -16,9 +17,10 @@ func Register(
 	workerConfig WorkerConfig,
 	queue *queue.Queue,
 	targetEncAlg crypto.EncryptionAlgorithm,
+	activeSigningKey GetActiveSigningWebKey,
 ) {
 	queue.ShouldStart()
-	queue.AddWorkers(NewWorker(workerConfig, targetEncAlg))
+	queue.AddWorkers(NewWorker(workerConfig, targetEncAlg, activeSigningKey, time.Now))
 }
 
 func Start(ctx context.Context) {

internal/execution/target/target.go

@@ -14,6 +14,15 @@ const (
 	TargetTypeAsync
 )
 
+type PayloadType uint
+
+const (
+	PayloadTypeUnspecified PayloadType = iota
+	PayloadTypeJSON
+	PayloadTypeJWT
+	PayloadTypeJWE
+)
+
 type Target struct {
 	ExecutionID      string              `json:"execution_id,omitempty"`
 	TargetID         string              `json:"target_id,omitempty"`
@@ -22,6 +31,9 @@ type Target struct {
 	Timeout          time.Duration       `json:"timeout,omitempty"`
 	InterruptOnError bool                `json:"interrupt_on_error,omitempty"`
 	SigningKey       *crypto.CryptoValue `json:"signing_key,omitempty"`
+	PayloadType      PayloadType         `json:"payload_type,omitempty"`
+	EncryptionKey    []byte              `json:"encryption_key,omitempty"`
+	EncryptionKeyID  string              `json:"encryption_key_id,omitempty"`
 }
 
 func (e *Target) GetExecutionID() string {
@@ -48,3 +60,13 @@ func (e *Target) GetSigningKey(alg crypto.EncryptionAlgorithm) (string, error) {
 	}
 	return crypto.DecryptString(e.SigningKey, alg)
 }
+func (e *Target) GetPayloadType() PayloadType {
+	return e.PayloadType
+}
+func (e *Target) GetEncryptionKey() []byte {
+	return e.EncryptionKey
+}
+
+func (e *Target) GetEncryptionKeyID() string {
+	return e.EncryptionKeyID
+}

internal/execution/target_test.go

@@ -1,85 +0,0 @@
-package execution
-
-import (
-	"encoding/json"
-	"io"
-	"net/http"
-	"net/http/httptest"
-	"reflect"
-	"time"
-)
-
-type testServer struct {
-	server *httptest.Server
-	called bool
-}
-
-func (s *testServer) URL() string {
-	return s.server.URL
-}
-
-func (s *testServer) Close() {
-	s.server.Close()
-}
-
-func (s *testServer) Called() bool {
-	return s.called
-}
-
-func testServerCall(
-	reqBody interface{},
-	sleep time.Duration,
-	statusCode int,
-	respBody interface{},
-) (string, func(), func() bool) {
-	server := &testServer{
-		called: false,
-	}
-
-	handler := func(w http.ResponseWriter, r *http.Request) {
-		server.called = true
-		if reqBody != nil {
-			data, err := json.Marshal(reqBody)
-			if err != nil {
-				http.Error(w, "error, marshall: "+err.Error(), http.StatusInternalServerError)
-				return
-			}
-			sentBody, err := io.ReadAll(r.Body)
-			if err != nil {
-				http.Error(w, "error, read body: "+err.Error(), http.StatusInternalServerError)
-				return
-			}
-			if !reflect.DeepEqual(data, sentBody) {
-				http.Error(w, "error, equal:\n"+string(data)+"\nsent:\n"+string(sentBody), http.StatusInternalServerError)
-				return
-			}
-		}
-		if statusCode != http.StatusOK {
-			http.Error(w, "error, statusCode", statusCode)
-			return
-		}
-
-		time.Sleep(sleep)
-
-		if respBody != nil {
-			w.Header().Set("Content-Type", "application/json")
-			resp, err := json.Marshal(respBody)
-			if err != nil {
-				http.Error(w, "error", http.StatusInternalServerError)
-				return
-			}
-			if _, err := w.Write(resp); err != nil {
-				http.Error(w, "error", http.StatusInternalServerError)
-				return
-			}
-		} else {
-			if _, err := io.WriteString(w, "finished successfully"); err != nil {
-				http.Error(w, "error", http.StatusInternalServerError)
-				return
-			}
-		}
-	}
-
-	server.server = httptest.NewServer(http.HandlerFunc(handler))
-	return server.URL(), server.Close, server.Called
-}

internal/execution/worker.go

@@ -18,9 +18,10 @@ type Worker struct {
 	river.WorkerDefaults[*exec_repo.Request]
 
 	config WorkerConfig
-	now    nowFunc
+	now    NowFunc
 
-	targetEncAlg crypto.EncryptionAlgorithm
+	targetEncAlg     crypto.EncryptionAlgorithm
+	activeSigningKey GetActiveSigningWebKey
 }
 
 // Timeout implements the Timeout-function of [river.Worker].
@@ -45,7 +46,7 @@ func (w *Worker) Work(ctx context.Context, job *river.Job[*exec_repo.Request]) e
 		return river.JobCancel(fmt.Errorf("unable to unmarshal targets because %w", err))
 	}
 
-	_, err = CallTargets(ctx, targets, exec_repo.ContextInfoFromRequest(job.Args), w.targetEncAlg)
+	_, err = CallTargets(ctx, targets, exec_repo.ContextInfoFromRequest(job.Args), w.targetEncAlg, w.activeSigningKey)
 	if err != nil {
 		// If there is an error returned from the targets, it means that the execution was interrupted
 		return river.JobCancel(fmt.Errorf("interruption during call of targets because %w", err))
@@ -53,8 +54,8 @@ func (w *Worker) Work(ctx context.Context, job *river.Job[*exec_repo.Request]) e
 	return nil
 }
 
-// nowFunc makes [time.Now] mockable
-type nowFunc func() time.Time
+// NowFunc makes [time.Now] mockable
+type NowFunc func() time.Time
 
 type WorkerConfig struct {
 	Workers             uint8
@@ -65,11 +66,14 @@ type WorkerConfig struct {
 func NewWorker(
 	config WorkerConfig,
 	targetEncAlg crypto.EncryptionAlgorithm,
+	activeSigningKey GetActiveSigningWebKey,
+	now NowFunc,
 ) *Worker {
 	return &Worker{
-		config:       config,
-		now:          time.Now,
-		targetEncAlg: targetEncAlg,
+		config:           config,
+		now:              now,
+		targetEncAlg:     targetEncAlg,
+		activeSigningKey: activeSigningKey,
 	}
 }
 

internal/execution/worker_test.go

@@ -1,4 +1,4 @@
-package execution
+package execution_test
 
 import (
 	"context"
@@ -15,6 +15,7 @@ import (
 
 	"github.com/zitadel/zitadel/internal/api/authz"
 	"github.com/zitadel/zitadel/internal/eventstore"
+	"github.com/zitadel/zitadel/internal/execution"
 	target_domain "github.com/zitadel/zitadel/internal/execution/target"
 	"github.com/zitadel/zitadel/internal/repository/action"
 	exec_repo "github.com/zitadel/zitadel/internal/repository/execution"
@@ -23,26 +24,44 @@ import (
 )
 
 type fieldsWorker struct {
-	now nowFunc
+	now execution.NowFunc
 }
 type argsWorker struct {
 	job *river.Job[*exec_repo.Request]
 }
 type wantWorker struct {
-	targets        []target_domain.Target
+	targets        []target
 	sendStatusCode int
 	err            assert.ErrorAssertionFunc
 }
 
-func newExecutionWorker(f fieldsWorker) *Worker {
-	return &Worker{
-		config: WorkerConfig{
+type target target_domain.Target
+
+func (t *target) validate(expectedBody []byte) func(*testing.T, []byte) bool {
+	switch t.PayloadType {
+	case target_domain.PayloadTypeUnspecified,
+		target_domain.PayloadTypeJSON:
+		return validateJSONPayload(expectedBody)
+	case target_domain.PayloadTypeJWT:
+		return validateJWTPayload(expectedBody)
+	case target_domain.PayloadTypeJWE:
+		return validateJWEPayload(expectedBody)
+	default:
+		return validateJSONPayload(expectedBody)
+	}
+}
+
+func newExecutionWorker(f fieldsWorker) *execution.Worker {
+	return execution.NewWorker(
+		execution.WorkerConfig{
 			Workers:             1,
 			TransactionDuration: 5 * time.Second,
 			MaxTtl:              5 * time.Minute,
 		},
-		now: f.now,
-	}
+		nil,
+		mockGetActiveSigningWebKey,
+		f.now,
+	)
 }
 
 const (
@@ -85,7 +104,7 @@ func Test_handleEventExecution(t *testing.T) {
 						},
 					},
 					wantWorker{
-						targets:        mockTargets(1),
+						targets:        mockTargets(target_domain.PayloadTypeJSON),
 						sendStatusCode: http.StatusOK,
 						err: func(tt assert.TestingT, err error, i ...interface{}) bool {
 							return errors.Is(err, new(river.JobCancelError))
@@ -153,7 +172,7 @@ func Test_handleEventExecution(t *testing.T) {
 						},
 					},
 					wantWorker{
-						targets:        mockTargets(1),
+						targets:        mockTargets(target_domain.PayloadTypeJSON),
 						sendStatusCode: http.StatusOK,
 						err:            nil,
 					}
@@ -187,7 +206,7 @@ func Test_handleEventExecution(t *testing.T) {
 						},
 					},
 					wantWorker{
-						targets:        mockTargets(1),
+						targets:        mockTargets(target_domain.PayloadTypeJSON),
 						sendStatusCode: http.StatusBadRequest,
 						err: func(tt assert.TestingT, err error, i ...interface{}) bool {
 							return errors.Is(err, zerrors.ThrowPreconditionFailed(nil, "EXEC-dra6yamk98", "Errors.Execution.Failed"))
@@ -223,7 +242,7 @@ func Test_handleEventExecution(t *testing.T) {
 						},
 					},
 					wantWorker{
-						targets:        mockTargets(3),
+						targets:        mockTargets(target_domain.PayloadTypeJSON, target_domain.PayloadTypeJWT, target_domain.PayloadTypeJWE),
 						sendStatusCode: http.StatusOK,
 						err:            nil,
 					}
@@ -234,15 +253,19 @@ func Test_handleEventExecution(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			f, a, w := tt.test()
 
+			body, err := json.Marshal(exec_repo.ContextInfoFromRequest(a.job.Args))
+			require.NoError(t, err)
+
 			closeFuncs := make([]func(), len(w.targets))
 			calledFuncs := make([]func() bool, len(w.targets))
 			for i := range w.targets {
-				url, closeF, calledF := testServerCall(
-					exec_repo.ContextInfoFromRequest(a.job.Args),
-					time.Second,
-					w.sendStatusCode,
-					nil,
-				)
+				url, closeF, calledF := listen(t, &callTestServer{
+					method:      http.MethodPost,
+					expectBody:  w.targets[i].validate(body),
+					timeout:     time.Second,
+					statusCode:  w.sendStatusCode,
+					respondBody: nil,
+				})
 				w.targets[i].Endpoint = url
 				closeFuncs[i] = closeF
 				calledFuncs[i] = calledF
@@ -273,24 +296,24 @@ func Test_handleEventExecution(t *testing.T) {
 	}
 }
 
-func mockTarget() target_domain.Target {
-	return target_domain.Target{
+func mockTarget(payloadType target_domain.PayloadType) target {
+	return target{
 		ExecutionID:      "executionID",
 		TargetID:         "targetID",
 		TargetType:       target_domain.TargetTypeWebhook,
 		Endpoint:         "endpoint",
 		Timeout:          time.Minute,
 		InterruptOnError: true,
+		PayloadType:      payloadType,
+		EncryptionKey:    encryptionKey,
+		EncryptionKeyID:  encryptionKeyID,
 	}
 }
 
-func mockTargets(count int) []target_domain.Target {
-	var targets []target_domain.Target
-	if count > 0 {
-		targets = make([]target_domain.Target, count)
-		for i := range targets {
-			targets[i] = mockTarget()
-		}
+func mockTargets(payloadTypes ...target_domain.PayloadType) []target {
+	targets := make([]target, len(payloadTypes))
+	for i, payloadType := range payloadTypes {
+		targets[i] = mockTarget(payloadType)
 	}
 	return targets
 }