zitadel/internal/execution/execution.go

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"
	"github.com/zitadel/zitadel/internal/repository/execution"
	"github.com/zitadel/zitadel/internal/telemetry/tracing"
	"github.com/zitadel/zitadel/internal/zerrors"
	"github.com/zitadel/zitadel/pkg/actions"
)

type ContextInfo interface {
	GetHTTPRequestBody() []byte
	GetContent() 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,
	client *http.Client,
) (_ any, 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, signerOnce, encrypters, client)
		// 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
		}
		if len(resp) > 0 {
			// error in unmarshalling
			if err := info.SetHTTPResponseBody(resp); err != nil && target.IsInterruptOnError() {
				return nil, err
			}
		}
	}
	return info.GetContent(), nil
}

type ContextInfoRequest interface {
	GetHTTPRequestBody() []byte
}

// CallTarget call the desired type of target with handling of responses
func CallTarget(
	ctx context.Context,
	target target_domain.Target,
	info ContextInfoRequest,
	alg crypto.EncryptionAlgorithm,
	signerOnce sign.SignerFunc,
	encrypters *sync.Map,
	client *http.Client,
) (res []byte, err error) {
	ctx, span := tracing.NewSpan(ctx)
	defer func() { span.EndWithError(err) }()

	signingKey, err := target.GetSigningKey(alg)
	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(), body, signingKey, client)
	// get request, return response and error
	case target_domain.TargetTypeCall:
		return Call(ctx, target.GetEndpoint(), target.GetTimeout(), body, signingKey, client)
	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, client); err != nil {
				logging.WithFields("target", target.GetTargetID()).OnError(err).Info(err)
			}
		}(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, client *http.Client) error {
	_, err := Call(ctx, url, timeout, body, signingKey, client)
	return err
}

// Call function to do a post HTTP request to a desired url with timeout
func Call(ctx context.Context, url string, timeout time.Duration, body []byte, signingKey string, client *http.Client) (_ []byte, err error) {
	ctx, cancel := context.WithTimeout(ctx, timeout)
	ctx, span := tracing.NewSpan(ctx)
	defer func() {
		cancel()
		span.EndWithError(err)
	}()

	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBuffer(body))
	if err != nil {
		return nil, err
	}
	req.Header.Set("Content-Type", "application/json")
	if signingKey != "" {
		req.Header.Set(actions.SigningHeader, actions.ComputeSignatureHeader(time.Now(), body, signingKey))
	}

	resp, err := client.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	return HandleResponse(resp)
}

func HandleResponse(resp *http.Response) ([]byte, error) {
	data, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}
	// Check for success between 200 and 299, redirect 300 to 399 is handled by the client, return error with statusCode >= 400
	if resp.StatusCode >= 200 && resp.StatusCode <= 299 {
		var errorBody ErrorBody
		if err := json.Unmarshal(data, &errorBody); err != nil {
			// if json unmarshal fails, body has no ErrorBody information, so will be taken as successful response
			return data, nil
		}
		if errorBody.ForwardedStatusCode != 0 || errorBody.ForwardedErrorMessage != "" {
			if errorBody.ForwardedStatusCode >= 400 && errorBody.ForwardedStatusCode < 500 {
				return nil, zhttp.HTTPStatusCodeToZitadelError(nil, errorBody.ForwardedStatusCode, "EXEC-reUaUZCzCp", errorBody.ForwardedErrorMessage)
			}
			return nil, zerrors.ThrowPreconditionFailed(nil, "EXEC-bmhNhpcqpF", errorBody.ForwardedErrorMessage)
		}
		// no ErrorBody filled in response, so will be taken as successful response
		return data, nil
	}

	return nil, zerrors.ThrowPreconditionFailed(nil, "EXEC-dra6yamk98", "Errors.Execution.Failed")
}

type ErrorBody struct {
	ForwardedStatusCode   int    `json:"forwardedStatusCode,omitempty"`
	ForwardedErrorMessage string `json:"forwardedErrorMessage,omitempty"`
}

func QueryExecutionTargetsForRequest(
	ctx context.Context,
	fullMethod string,
) []target_domain.Target {
	ctx, span := tracing.NewSpan(ctx)
	defer span.End()

	requestTargets, _ := authz.GetInstance(ctx).ExecutionRouter().GetEventBestMatch(execution.ID(domain.ExecutionTypeRequest, fullMethod))
	return requestTargets
}

func QueryExecutionTargetsForResponse(
	ctx context.Context,
	fullMethod string,
) []target_domain.Target {
	ctx, span := tracing.NewSpan(ctx)
	defer span.End()

	responseTargets, _ := authz.GetInstance(ctx).ExecutionRouter().GetEventBestMatch(execution.ID(domain.ExecutionTypeResponse, fullMethod))
	return responseTargets
}

func QueryExecutionTargetsForFunction(ctx context.Context, function string) []target_domain.Target {
	executionTargets, _ := authz.GetInstance(ctx).ExecutionRouter().GetEventBestMatch(function)
	return executionTargets
}