From 9a29c6e1b27fee55e813ca813bb9e489a67c9e37 Mon Sep 17 00:00:00 2001 From: Denis Arh Date: Tue, 20 Jul 2021 18:37:23 +0200 Subject: [PATCH] Fix deferred workflow execution This fix addresses execution of deferred workflows and access control when setting run-as user. Some additional standardisation on how we write mutex/lock code. --- automation/service/service.go | 2 +- automation/service/trigger.go | 2 +- automation/service/workflow.go | 22 +++++++++++++--------- pkg/scheduler/helpers.go | 4 ++++ pkg/scheduler/service.go | 19 ++++++++++++------- 5 files changed, 31 insertions(+), 18 deletions(-) diff --git a/automation/service/service.go b/automation/service/service.go index 01beed031..be6a1f957 100644 --- a/automation/service/service.go +++ b/automation/service/service.go @@ -28,7 +28,7 @@ type ( } userService interface { - FindByID(ctx context.Context, userID uint64) (*types.User, error) + FindByAny(ctx context.Context, identifier interface{}) (*types.User, error) } ) diff --git a/automation/service/trigger.go b/automation/service/trigger.go index bc81b577c..3b33ce476 100644 --- a/automation/service/trigger.go +++ b/automation/service/trigger.go @@ -465,7 +465,7 @@ func (svc *trigger) registerWorkflow(ctx context.Context, wf *types.Workflow, tt } if wf.RunAs > 0 { - if runAs, err = DefaultUser.FindByID(ctx, wf.RunAs); err != nil { + if runAs, err = DefaultUser.FindByAny(ctx, wf.RunAs); err != nil { return fmt.Errorf("failed to load run-as user %d: %w", wf.RunAs, err) } else if !runAs.Valid() { return fmt.Errorf("invalid user %d used for workflow run-as", wf.RunAs) diff --git a/automation/service/workflow.go b/automation/service/workflow.go index b58a426da..2ccb67674 100644 --- a/automation/service/workflow.go +++ b/automation/service/workflow.go @@ -5,8 +5,6 @@ import ( "reflect" "sync" - "github.com/cortezaproject/corteza-server/pkg/options" - "github.com/cortezaproject/corteza-server/automation/types" "github.com/cortezaproject/corteza-server/pkg/actionlog" intAuth "github.com/cortezaproject/corteza-server/pkg/auth" @@ -16,6 +14,7 @@ import ( "github.com/cortezaproject/corteza-server/pkg/filter" "github.com/cortezaproject/corteza-server/pkg/handle" "github.com/cortezaproject/corteza-server/pkg/label" + "github.com/cortezaproject/corteza-server/pkg/options" "github.com/cortezaproject/corteza-server/pkg/rbac" "github.com/cortezaproject/corteza-server/pkg/wfexec" "github.com/cortezaproject/corteza-server/store" @@ -570,7 +569,7 @@ func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.Workfl ssp.Input = scope.MustMerge(p.Input) if wf.RunAs > 0 { - if runAs, err = DefaultUser.FindByID(ctx, wf.RunAs); err != nil { + if runAs, err = DefaultUser.FindByAny(ctx, wf.RunAs); err != nil { return } } @@ -604,12 +603,8 @@ func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.Workfl return results, stacktrace, svc.recordAction(ctx, wap, WorkflowActionExecute, err) } -func makeWorkflowHandler(ac workflowExecController, s *session, t *types.Trigger, wf *types.Workflow, g *wfexec.Graph, runAs intAuth.Identifiable) eventbus.HandlerFn { +func makeWorkflowHandler(ac workflowExecController, s *session, t *types.Trigger, wf *types.Workflow, g *wfexec.Graph, _runAs intAuth.Identifiable) eventbus.HandlerFn { return func(ctx context.Context, ev eventbus.Event) (err error) { - if !ac.CanExecuteWorkflow(ctx, wf) { - return WorkflowErrNotAllowedToExecute() - } - var ( // create session scope from predefined workflow scope and trigger input scope = wf.Scope.MustMerge(t.Input) @@ -618,7 +613,7 @@ func makeWorkflowHandler(ac workflowExecController, s *session, t *types.Trigger // The returned closure needs to have its own instance, so it doesn't // affect the instance bound to the workflow handler - runAs = runAs + runAs = _runAs ) if enc, is := ev.(varsEncoder); is { @@ -638,6 +633,15 @@ func makeWorkflowHandler(ac workflowExecController, s *session, t *types.Trigger // - use http auth header and get username // - use from/to/replyTo and use that as an identifier runAs = intAuth.GetIdentityFromContext(ctx) + } else { + // Running workflow with a different security context + ctx = intAuth.SetIdentityToContext(ctx, runAs) + } + + // User (either invoker or one set in the security descriptor) MUST have + // permissions to execute this workflow + if !ac.CanExecuteWorkflow(ctx, wf) { + return WorkflowErrNotAllowedToExecute() } wait, err = s.Start(g, runAs, types.SessionStartParams{ diff --git a/pkg/scheduler/helpers.go b/pkg/scheduler/helpers.go index 2b2b07d3e..b52611af7 100644 --- a/pkg/scheduler/helpers.go +++ b/pkg/scheduler/helpers.go @@ -3,14 +3,17 @@ package scheduler import ( "time" + "github.com/cortezaproject/corteza-server/pkg/logger" "github.com/getsentry/sentry-go" "github.com/gorhill/cronexpr" + "go.uber.org/zap" ) // OnInterval parses all given strings as crontab expressions (ii) and returns true if any of them matches current time func OnInterval(ii ...string) bool { match, err := onInterval(now(), ii...) if err != nil { + logger.Default().Error("failed to parse interval value", zap.Strings("value", ii), zap.Error(err)) sentry.CaptureException(err) } return match @@ -44,6 +47,7 @@ func onInterval(now time.Time, ii ...string) (bool, error) { func OnTimestamp(tt ...string) bool { match, err := onTimestamp(now(), tt...) if err != nil { + logger.Default().Error("failed to parse timestamp value", zap.Strings("value", tt), zap.Error(err)) sentry.CaptureException(err) } return match diff --git a/pkg/scheduler/service.go b/pkg/scheduler/service.go index 098b3b760..0add63665 100644 --- a/pkg/scheduler/service.go +++ b/pkg/scheduler/service.go @@ -19,14 +19,14 @@ type ( dispatcher dispatcher // Read & write locking - l *sync.RWMutex + l sync.RWMutex // Simple chan to control if service is running or not ticker *time.Ticker } dispatcher interface { - Dispatch(ctx context.Context, ev eventbus.Event) + WaitFor(ctx context.Context, ev eventbus.Event) error } ) @@ -65,7 +65,7 @@ func NewService(log *zap.Logger, d dispatcher, interval time.Duration) *service } var svc = &service{ - l: &sync.RWMutex{}, + l: sync.RWMutex{}, log: log.Named("scheduler"), interval: interval, dispatcher: d, @@ -130,7 +130,7 @@ func (svc *service) Start(ctx context.Context) { }() } -func (svc service) watch(ctx context.Context) { +func (svc *service) watch(ctx context.Context) { defer sentry.Recover() defer func() { defer svc.log.Debug("stopped") @@ -155,14 +155,14 @@ func (svc service) watch(ctx context.Context) { } } -func (svc service) Started() (started bool) { +func (svc *service) Started() (started bool) { svc.l.RLock() defer svc.l.RUnlock() return svc.ticker != nil } -func (svc service) dispatch(ctx context.Context) { +func (svc *service) dispatch(ctx context.Context) { svc.l.RLock() ee := make([]eventbus.Event, len(svc.events)) @@ -173,6 +173,11 @@ func (svc service) dispatch(ctx context.Context) { defer svc.l.RUnlock() for _, ev := range ee { - go svc.dispatcher.Dispatch(ctx, ev) + go func(ev eventbus.Event) { + err := svc.dispatcher.WaitFor(ctx, ev) + if err != nil { + svc.log.Warn("failed to execute scheduled trigger", zap.Error(err)) + } + }(ev) } }