mirror of
https://github.com/andwati/zitadel.git
synced 2026-10-18 13:52:51 +00:00
Marco A.andGitHub
b2532e9666
* Inject DenyList from config to `StartCommands()` * Implementation draft * Move address checker to separate package * Migrate usages of actions.AddressChecker to denylist.AddressChecker * Rename denylist package files * Pass []denylist.AddressChecker to StartCommand * Add DenyList to defaults.yaml and add custom config parser * net: add HostnameToIPList function * denylist: Add IsHostBlocked() * actions: use denylist.IsHostBlocked() * command: Inject ip lookup function + extend add target validationt test * command: Unexport ChangeTarget.IsValid() * command: Add denyList check on ChangeTarget validation * command: Export ActionsV2DenyList and IPLookupFunction params * Lint fix * Check denylist during action execution * Fix integration tests * Apply suggestions * Add `mapstructure.StringToSliceHookFunc()` to decode `HTTPConfigDecodeHook`
98 lines
2.9 KiB
Go
98 lines
2.9 KiB
Go
package execution
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/riverqueue/river"
|
|
|
|
"github.com/zitadel/zitadel/internal/crypto"
|
|
"github.com/zitadel/zitadel/internal/denylist"
|
|
target_domain "github.com/zitadel/zitadel/internal/execution/target"
|
|
exec_repo "github.com/zitadel/zitadel/internal/repository/execution"
|
|
)
|
|
|
|
type Worker struct {
|
|
river.WorkerDefaults[*exec_repo.Request]
|
|
|
|
config WorkerConfig
|
|
now NowFunc
|
|
|
|
targetEncAlg crypto.EncryptionAlgorithm
|
|
activeSigningKey GetActiveSigningWebKey
|
|
}
|
|
|
|
// Timeout implements the Timeout-function of [river.Worker].
|
|
// Maximum time a job can run before the context gets canceled.
|
|
// The time can be shorter than the sum of target timeouts, this is expected behavior to not block the request indefinitely.
|
|
func (w *Worker) Timeout(*river.Job[*exec_repo.Request]) time.Duration {
|
|
return w.config.TransactionDuration
|
|
}
|
|
|
|
// Work implements [river.Worker].
|
|
func (w *Worker) Work(ctx context.Context, job *river.Job[*exec_repo.Request]) error {
|
|
ctx = ContextWithExecuter(ctx, job.Args.Aggregate)
|
|
|
|
// if the event is too old, we can directly return as it will be removed anyway
|
|
if job.CreatedAt.Add(w.config.MaxTtl).Before(w.now()) {
|
|
return river.JobCancel(errors.New("event is too old"))
|
|
}
|
|
|
|
targets, err := TargetsFromRequest(job.Args)
|
|
if err != nil {
|
|
// If we are not able to get the targets from the request, we can cancel the job, as we have nothing to call
|
|
return river.JobCancel(fmt.Errorf("unable to unmarshal targets because %w", err))
|
|
}
|
|
|
|
_, err = CallTargets(ctx, targets, exec_repo.ContextInfoFromRequest(job.Args), w.targetEncAlg, w.activeSigningKey, w.config.DenyList)
|
|
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))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// NowFunc makes [time.Now] mockable
|
|
type NowFunc func() time.Time
|
|
|
|
type WorkerConfig struct {
|
|
Workers uint8
|
|
TransactionDuration time.Duration
|
|
MaxTtl time.Duration
|
|
DenyList []denylist.AddressChecker
|
|
}
|
|
|
|
func NewWorker(
|
|
config WorkerConfig,
|
|
targetEncAlg crypto.EncryptionAlgorithm,
|
|
activeSigningKey GetActiveSigningWebKey,
|
|
now NowFunc,
|
|
) *Worker {
|
|
return &Worker{
|
|
config: config,
|
|
now: now,
|
|
targetEncAlg: targetEncAlg,
|
|
activeSigningKey: activeSigningKey,
|
|
}
|
|
}
|
|
|
|
var _ river.Worker[*exec_repo.Request] = (*Worker)(nil)
|
|
|
|
func (w *Worker) Register(workers *river.Workers, queues map[string]river.QueueConfig) {
|
|
river.AddWorker(workers, w)
|
|
queues[exec_repo.QueueName] = river.QueueConfig{
|
|
MaxWorkers: int(w.config.Workers),
|
|
}
|
|
}
|
|
|
|
func TargetsFromRequest(e *exec_repo.Request) ([]target_domain.Target, error) {
|
|
var targets []target_domain.Target
|
|
if err := json.Unmarshal(e.TargetsData, &targets); err != nil {
|
|
return nil, err
|
|
}
|
|
return targets, nil
|
|
}
|