DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Fix deferred workflow execution

Browse Source 
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.
This commit is contained in:
Denis Arh
2021-07-21 12:02:01 +02:00
parent 774354d663
commit 9a29c6e1b2
5 changed files with 31 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
automation/service/service.go
+1 -1
View File
@@ -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)
}
)
automation/service/trigger.go
+1 -1
View File
@@ -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)
automation/service/workflow.go
+13 -9
View File
@@ -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{
pkg/scheduler/helpers.go
+4
View File
@@ -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
pkg/scheduler/service.go
+12 -7
View File
@@ -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)
}
}