diff --git a/automation/service/session.go b/automation/service/session.go index d833222f8..8d7d69336 100644 --- a/automation/service/session.go +++ b/automation/service/session.go @@ -38,6 +38,8 @@ type ( workflowID uint64 session chan *wfexec.Session graph *wfexec.Graph + invoker auth.Identifiable + runner auth.Identifiable trace bool callStack []uint64 } @@ -169,14 +171,36 @@ func (svc *session) PendingPrompts(ctx context.Context) (pp []*wfexec.PendingPro // // Start is an asynchronous operation // +// Please note that context passed to the function is NOT the the one that is +// used for the execution of the workflow. See watch function! +// // It does not check user's permissions to execute workflow(s) so it should be used only when ! -func (svc *session) Start(g *wfexec.Graph, i auth.Identifiable, ssp types.SessionStartParams) (wait WaitFn, err error) { +func (svc *session) Start(ctx context.Context, g *wfexec.Graph, ssp types.SessionStartParams) (wait WaitFn, err error) { var ( start wfexec.Step ) + if g == nil { + return nil, errors.InvalidData("cannot start workflow, uninitialized graph") + } + + if len(ssp.CallStack) > svc.opt.CallStackSize { + return nil, WorkflowErrMaximumCallStackSizeExceeded() + } + + ssp.CallStack = append(ssp.CallStack, ssp.WorkflowID) + + if ssp.Invoker == nil { + return nil, errors.InvalidData("cannot start workflow without user") + } + + if ssp.Runner == nil { + ssp.Runner = ssp.Invoker + } + if ssp.StepID == 0 { - // starting step is not explicitly workflows on trigger, find orphan step + // starting step is not explicitly set + // find orphan step switch oo := g.Orphans(); len(oo) { case 1: start = oo[0] @@ -192,15 +216,19 @@ func (svc *session) Start(g *wfexec.Graph, i auth.Identifiable, ssp types.Sessio } var ( - ctx = auth.SetIdentityToContext(context.Background(), i) - ses = svc.spawn(g, ssp.WorkflowID, ssp.Trace, ssp.CallStack) + ses = svc.spawn(g, ssp.WorkflowID, ssp.Trace, ssp.CallStack, ssp.Runner, ssp.Invoker) ) ses.CreatedAt = *now() - ses.CreatedBy = i.Identity() + ses.CreatedBy = ssp.Invoker.Identity() ses.Status = types.SessionStarted ses.Apply(ssp) + _ = ssp.Input.AssignFieldValue("eventType", expr.Must(expr.NewString(ssp.EventType))) + _ = ssp.Input.AssignFieldValue("resourceType", expr.Must(expr.NewString(ssp.ResourceType))) + _ = ssp.Input.AssignFieldValue("invoker", expr.Must(expr.NewAny(ssp.Invoker))) + _ = ssp.Input.AssignFieldValue("runner", expr.Must(expr.NewAny(ssp.Runner))) + if err = ses.Exec(ctx, start, ssp.Input); err != nil { return } @@ -241,13 +269,15 @@ func (svc *session) Resume(sessionID, stateID uint64, i auth.Identifiable, input // // We need initial context for the session because we want to catch all cancellations or timeouts from there // and not from any potential HTTP requests or similar temporary context that can prematurely destroy a workflow session -func (svc *session) spawn(g *wfexec.Graph, workflowID uint64, trace bool, callStack []uint64) (ses *types.Session) { +func (svc *session) spawn(g *wfexec.Graph, workflowID uint64, trace bool, callStack []uint64, invoker, runner auth.Identifiable) (ses *types.Session) { s := &spawn{ workflowID: workflowID, session: make(chan *wfexec.Session, 1), graph: g, trace: trace, callStack: callStack, + invoker: invoker, + runner: runner, } // Send new-session request @@ -262,6 +292,7 @@ func (svc *session) spawn(g *wfexec.Graph, workflowID uint64, trace bool, callSt return ses } +// Watch looks over session's spawn queue func (svc *session) Watch(ctx context.Context) { gcTicker := time.NewTicker(time.Second) lpTicker := time.NewTicker(time.Second * 30) @@ -276,6 +307,8 @@ func (svc *session) Watch(ctx context.Context) { case <-ctx.Done(): return case s := <-svc.spawnQueue: + var execCtx = context.Background() + opts := []wfexec.SessionOpt{ wfexec.SetWorkflowID(s.workflowID), wfexec.SetCallStack(s.callStack...), @@ -290,7 +323,15 @@ func (svc *session) Watch(ctx context.Context) { ) } - s.session <- wfexec.NewSession(ctx, s.graph, opts...) + // Encode runner into execution context + // runner is used as identity and for access control + execCtx = auth.SetIdentityToContext(execCtx, s.runner) + + // Encode invoker into execution context + // invoker is used + execCtx = context.WithValue(execCtx, workflowInvokerCtxKey{}, s.invoker) + + s.session <- wfexec.NewSession(execCtx, s.graph, opts...) // case time for a pool cleanup // @todo cleanup pool when sessions are complete @@ -370,6 +411,8 @@ func (svc *session) stateChangeHandler(ctx context.Context) wfexec.StateChangeHa return } + log = log.With(zap.Uint64("workflowID", ses.WorkflowID)) + var ( // By default, we want to update session when new status is prompted, delayed, completed or failed // But if status is active, we'll flush it every X frames (sessionStateFlushFrequency) diff --git a/automation/service/session_test.go b/automation/service/session_test.go index 712ffed68..a2f97260d 100644 --- a/automation/service/session_test.go +++ b/automation/service/session_test.go @@ -1,9 +1,11 @@ package service import ( + "context" "testing" "github.com/cortezaproject/corteza-server/automation/types" + "github.com/cortezaproject/corteza-server/pkg/auth" "github.com/cortezaproject/corteza-server/pkg/wfexec" "github.com/stretchr/testify/require" ) @@ -13,19 +15,21 @@ func TestSession_Start(t *testing.T) { req = require.New(t) ses = &session{} g = wfexec.NewGraph() + ctx = context.Background() + err error ) - _, err = ses.Start(g, nil, types.SessionStartParams{}) + _, err = ses.Start(ctx, g, types.SessionStartParams{Invoker: auth.Anonymous()}) req.EqualError(err, "could not find starting step") g.AddStep(wfexec.NewGenericStep(nil)) - _, err = ses.Start(g, nil, types.SessionStartParams{StepID: 4321}) + _, err = ses.Start(ctx, g, types.SessionStartParams{StepID: 4321, Invoker: auth.Anonymous()}) req.EqualError(err, "trigger staring step references non-existing step") // Adding another orphaned step and starting session w/o explicitly specifying the starting step g.AddStep(wfexec.NewGenericStep(nil)) - _, err = ses.Start(g, nil, types.SessionStartParams{}) + _, err = ses.Start(ctx, g, types.SessionStartParams{Invoker: auth.Anonymous()}) req.EqualError(err, "cannot start workflow session multiple starting steps found") // add a generic step with a known ID so we can use it as a starting point @@ -34,7 +38,7 @@ func TestSession_Start(t *testing.T) { g.AddStep(s) // add parents to the 42 step g.AddStep(wfexec.NewGenericStep(nil), s) - _, err = ses.Start(g, nil, types.SessionStartParams{StepID: 42}) + _, err = ses.Start(ctx, g, types.SessionStartParams{StepID: 42, Invoker: auth.Anonymous()}) req.EqualError(err, "cannot start workflow on a step with parents") } diff --git a/automation/service/trigger.go b/automation/service/trigger.go index 7c252bf6e..1d7f80ef6 100644 --- a/automation/service/trigger.go +++ b/automation/service/trigger.go @@ -528,11 +528,15 @@ func (svc *trigger) registerTriggers(wf *types.Workflow, runAs auth.Identifiable registerWorkflow = wf.Enabled && wf.DeletedAt == nil ) - // convert only registerable and issuless workflwos + // convert only registrable and workflows without issues if registerWorkflow && len(wf.Issues) == 0 { // Convert workflow only when valid (no issues, enable, not delete) if g, issues = Convert(svc.workflow, wf); len(issues) > 0 { wfLog.Error("failed to convert workflow to graph", zap.Error(issues)) + _ = issues.Walk(func(i *types.WorkflowIssue) error { + wfLog.Debug("workflow issue found: "+i.Description, zap.Any("culprit", i.Culprit)) + return nil + }) g = nil } } @@ -577,7 +581,7 @@ func (svc *trigger) registerTriggers(wf *types.Workflow, runAs auth.Identifiable ).Wrap(wf.Issues) } } else { - handlerFn = makeWorkflowHandler(svc.ac, svc.session, t, wf, g, runAs) + handlerFn = makeWorkflowHandler(svc.workflow, wf, t) } ops = append( diff --git a/automation/service/workflow.go b/automation/service/workflow.go index e05837578..7f56aa10d 100644 --- a/automation/service/workflow.go +++ b/automation/service/workflow.go @@ -2,6 +2,7 @@ package service import ( "context" + "fmt" "reflect" "sync" @@ -18,8 +19,6 @@ import ( "github.com/cortezaproject/corteza-server/pkg/rbac" "github.com/cortezaproject/corteza-server/pkg/wfexec" "github.com/cortezaproject/corteza-server/store" - sysAutoTypes "github.com/cortezaproject/corteza-server/system/automation" - sysTypes "github.com/cortezaproject/corteza-server/system/types" "go.uber.org/zap" ) @@ -36,8 +35,11 @@ type ( log *zap.Logger - // maps resolved workflow graphs to workflow ID (key, uint64) - wfgs map[uint64]*wfexec.Graph + // cache of workflows, graphs to workflow ID (key, uint64) + cache map[uint64]*wfCacheItem + + // handle to workflow index + wIndex map[string]uint64 // workflow function registry reg *registry @@ -47,6 +49,16 @@ type ( parser expr.Parsable } + wfCacheItem struct { + wf *types.Workflow + + // caching exec graph + g *wfexec.Graph + + // caching user we'll executing workflow with + runAs intAuth.Identifiable + } + workflowAccessController interface { CanSearchWorkflows(context.Context) bool CanCreateWorkflow(context.Context) bool @@ -72,6 +84,8 @@ type ( workflowUpdateHandler func(ctx context.Context, ns *types.Workflow) (workflowChanges, error) workflowChanges uint8 + + workflowInvokerCtxKey struct{} ) const ( @@ -91,7 +105,8 @@ func Workflow(log *zap.Logger, corredorOpt options.CorredorOpt, opt options.Work triggers: DefaultTrigger, session: DefaultSession, eventbus: eventbus.Service(), - wfgs: make(map[uint64]*wfexec.Graph), + cache: make(map[uint64]*wfCacheItem), + wIndex: make(map[string]uint64), mux: &sync.RWMutex{}, parser: expr.NewParser(), reg: Registry(), @@ -180,6 +195,8 @@ func (svc *workflow) Create(ctx context.Context, new *types.Workflow) (wf *types var ( wap = &workflowActionProps{new: new} cUser = intAuth.GetIdentityFromContext(ctx).Identity() + g *wfexec.Graph + runAs intAuth.Identifiable ) err = store.Tx(ctx, svc.store, func(ctx context.Context, s store.Storer) (err error) { @@ -215,9 +232,15 @@ func (svc *workflow) Create(ctx context.Context, new *types.Workflow) (wf *types CreatedBy: cUser, } - if wf.Issues, err = svc.validateWorkflow(ctx, wf); err != nil { + if g, runAs, err = svc.validateWorkflow(ctx, wf); err != nil { return - } else if len(wf.Issues) == 0 { + } + + if err = svc.updateCache(wf, runAs, g); err != nil { + return + } + + if len(wf.Issues) == 0 { if err = svc.triggers.registerWorkflows(ctx, wf); err != nil { return err } @@ -256,34 +279,20 @@ func (svc *workflow) DeleteByID(ctx context.Context, workflowID uint64) error { return workflowUnchanged, err } - if res.Issues, err = svc.validateWorkflow(ctx, res); err != nil { - return workflowUnchanged, err - } else if len(res.Issues) == 0 { - if err := svc.triggers.registerWorkflows(ctx, res); err != nil { - return workflowUnchanged, err - } - } - return changes, err - })) } func (svc *workflow) UndeleteByID(ctx context.Context, workflowID uint64) error { return trim1st(svc.updater(ctx, workflowID, WorkflowActionUndelete, func(ctx context.Context, res *types.Workflow) (workflowChanges, error) { - changes, err := svc.handleUndelete(ctx, res) + var ( + changes, err = svc.handleUndelete(ctx, res) + ) + if err != nil { return workflowUnchanged, err } - if res.Issues, err = svc.validateWorkflow(ctx, res); err != nil { - return workflowUnchanged, err - } else if len(res.Issues) == 0 { - if err := svc.triggers.registerWorkflows(ctx, res); err != nil { - return workflowUnchanged, err - } - } - return changes, err })) @@ -305,6 +314,8 @@ func (svc *workflow) updater(ctx context.Context, workflowID uint64, action func res *types.Workflow aProps = &workflowActionProps{workflow: &types.Workflow{ID: workflowID}} err error + g *wfexec.Graph + runAs intAuth.Identifiable ) err = store.Tx(ctx, svc.store, func(ctx context.Context, s store.Storer) (err error) { @@ -324,9 +335,15 @@ func (svc *workflow) updater(ctx context.Context, workflowID uint64, action func return err } - if res.Issues, err = svc.validateWorkflow(ctx, res); err != nil { + if g, runAs, err = svc.validateWorkflow(ctx, res); err != nil { return - } else if len(res.Issues) == 0 { + } + + if err = svc.updateCache(res, runAs, g); err != nil { + return + } + + if len(res.Issues) == 0 { if err = svc.triggers.registerWorkflows(ctx, res); err != nil { return err } @@ -344,12 +361,16 @@ func (svc *workflow) updater(ctx context.Context, workflowID uint64, action func } } - return err + return }) return res, svc.recordAction(ctx, aProps, action, err) } +func (svc *workflow) handleToID(h string) uint64 { + return svc.wIndex[h] +} + func (svc workflow) handleUpdate(upd *types.Workflow) workflowUpdateHandler { return func(ctx context.Context, res *types.Workflow) (changes workflowChanges, err error) { if isStale(upd.UpdatedAt, res.UpdatedAt, res.CreatedAt) { @@ -472,60 +493,78 @@ func (svc workflow) handleUndelete(ctx context.Context, res *types.Workflow) (wo } func (svc *workflow) Load(ctx context.Context) error { - wwf, _, err := store.SearchAutomationWorkflows(ctx, svc.store, types.WorkflowFilter{ - Deleted: filter.StateInclusive, - Disabled: filter.StateExcluded, - }) + var ( + wwf, _, err = store.SearchAutomationWorkflows(ctx, svc.store, types.WorkflowFilter{ + Deleted: filter.StateInclusive, + Disabled: filter.StateExcluded, + }) + g *wfexec.Graph + runAs intAuth.Identifiable + ) if err != nil { return err } + _ = wwf.Walk(func(wf *types.Workflow) error { + svc.wIndex[wf.Handle] = wf.ID + + if g, runAs, err = svc.validateWorkflow(ctx, wf); err != nil { + return err + } + + if err = svc.updateCache(wf, runAs, g); err != nil { + return err + } + + return nil + }) + return svc.triggers.registerWorkflows(ctx, wwf...) } +// updateCache +func (svc *workflow) updateCache(wf *types.Workflow, runAs intAuth.Identifiable, g *wfexec.Graph) (err error) { + defer svc.mux.Unlock() + svc.mux.Lock() + if wf.Executable() { + svc.cache[wf.ID] = &wfCacheItem{g: g, wf: wf, runAs: runAs} + } else { + // remove deleted + delete(svc.cache, wf.ID) + } + + return +} + func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.WorkflowExecParams) (*expr.Vars, types.Stacktrace, error) { var ( - runAs intAuth.Identifiable wap = &workflowActionProps{} - wf *types.Workflow t *types.Trigger results *expr.Vars wait WaitFn stacktrace types.Stacktrace - - runner, invoker *sysTypes.User ) err := func() (err error) { - wf, err = loadWorkflow(ctx, svc.store, workflowID) - if err != nil { - return + svc.mux.Lock() + if nil == svc.cache[workflowID] || nil == svc.cache[workflowID].wf { + svc.mux.Unlock() + return WorkflowErrNotFound() } + wf := svc.cache[workflowID].wf + svc.mux.Unlock() + wap.setWorkflow(wf) - if !svc.ac.CanExecuteWorkflow(ctx, wf) { - return WorkflowErrNotAllowedToExecute() - } - - // User wants to trace workflow execution - // This means we'll allow him to specify any (orphaned) step - // even if it's not linked to onManual trigger - if p.Trace && !svc.ac.CanManageSessionsOnWorkflow(ctx, wf) { - return WorkflowErrNotAllowedToExecute() - } if !wf.Enabled && !p.Trace { return WorkflowErrDisabled() } - g, convErr := Convert(svc, wf) - if len(convErr) > 0 { - return convErr - } - // Find the trigger. + // @todo can we cache this as well? t, err = func() (*types.Trigger, error) { var tt types.TriggerSet // Load triggers directly from the store. At this point we do not care @@ -552,23 +591,6 @@ func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.Workfl return } - // Start with workflow scope - scope := wf.Scope.MustMerge() - - callStack := wfexec.GetContextCallStack(ctx) - if len(callStack) > svc.opt.CallStackSize { - return WorkflowErrMaximumCallStackSizeExceeded() - } - - ssp := types.SessionStartParams{ - WorkflowID: wf.ID, - KeepFor: wf.KeepSessions, - Trace: wf.Trace || p.Trace, - StepID: p.StepID, - - CallStack: callStack, - } - if !p.Trace { if t == nil { return WorkflowErrUnknownWorkflowStep() @@ -579,59 +601,19 @@ func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.Workfl if t != nil { wap.setTrigger(t) + p.StepID = t.StepID + p.EventType = t.EventType + p.ResourceType = t.ResourceType - // Add trigger's input to scope - scope = scope.MustMerge(t.Input) - _ = scope.AssignFieldValue("eventType", expr.Must(expr.NewString(t.EventType))) - _ = scope.AssignFieldValue("resourceType", expr.Must(expr.NewString(t.ResourceType))) - - ssp.StepID = t.StepID - ssp.EventType = t.EventType - ssp.ResourceType = t.ResourceType + // merge with input from trigger + // with trigger input vars are overwritten by input vars + p.Input = t.Input.MustMerge(p.Input) } else { - ssp.EventType = "onTrace" - ssp.ResourceType = "" + p.EventType = "onTrace" } - // Returns context with identity set to service user - // - // Current user (identity in the context) might not have - // sufficient privileges to load info about invoker and runner - sysUserCtx := func() context.Context { - return intAuth.SetIdentityToContext(ctx, intAuth.ServiceUser()) - } - - if invokerId := intAuth.GetIdentityFromContext(ctx).Identity(); invokerId > 0 { - var is bool - if invoker, is = intAuth.GetIdentityFromContext(ctx).(*sysTypes.User); !is { - if invoker, err = DefaultUser.FindByAny(sysUserCtx(), invokerId); err != nil { - return - } - } - - runner = invoker - } - - if wf.RunAs > 0 { - if runner, err = DefaultUser.FindByAny(sysUserCtx(), wf.RunAs); err != nil { - return - } - } - - if runAs == nil { - // Default to current user - runAs = intAuth.GetIdentityFromContext(ctx) - } - - // @todo find a better way to typify expression values - // so that we do not have to import automation types from the system component - _ = scope.AssignFieldValue("invoker", expr.Must(sysAutoTypes.NewUser(invoker))) - _ = scope.AssignFieldValue("runner", expr.Must(sysAutoTypes.NewUser(runner))) - - // Finally, assign input values - ssp.Input = scope.MustMerge(p.Input) - - wait, err = svc.session.Start(g, runAs, ssp) + //wait, err = svc.session.Start(g, ssp) + wait, err = svc.exec(ctx, wf, p) if err != nil { return err @@ -649,87 +631,118 @@ func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.Workfl // reuse scope for results // this will be decoded back to event properties results, _, stacktrace, err = wait(ctx) - return + return err }() return results, stacktrace, svc.recordAction(ctx, wap, WorkflowActionExecute, err) } // validates workflow by trying to convert it to graph and checking assigned triggers -func (svc *workflow) validateWorkflow(ctx context.Context, wf *types.Workflow) (types.WorkflowIssueSet, error) { - _, wis := Convert(svc, wf) +func (svc *workflow) validateWorkflow(ctx context.Context, wf *types.Workflow) (g *wfexec.Graph, runAs intAuth.Identifiable, err error) { + var ( + tt []*types.Trigger + ) - tt, _, err := store.SearchAutomationTriggers(ctx, svc.store, types.TriggerFilter{ + g, wf.Issues = Convert(svc, wf) + + tt, _, err = store.SearchAutomationTriggers(ctx, svc.store, types.TriggerFilter{ WorkflowID: types.WorkflowSet{wf}.IDs(), Deleted: filter.StateExcluded, Disabled: filter.StateExcluded, }) if err != nil { - return nil, err + return } - wis = append(wis, validateWorkflowTriggers(wf, tt...)...) + wf.Issues = append(wf.Issues, validateWorkflowTriggers(wf, tt...)...) - return wis, nil + // Returns context with identity set to service user + // + // Current user (identity in the context) might not have + // sufficient privileges to load info about invoker and runner + sysUserCtx := func() context.Context { + return intAuth.SetIdentityToContext(ctx, intAuth.ServiceUser()) + } + + // @todo this might not be the smartest thing, users might get invalidated after + // we add cache them as workflow runners + if wf.RunAs > 0 { + if runAs, err = DefaultUser.FindByAny(sysUserCtx(), wf.RunAs); err != nil { + wf.Issues = wf.Issues.Append(fmt.Errorf("failed to load run-as user %d: %w", wf.RunAs, err), nil) + } else if !runAs.Valid() { + wf.Issues = wf.Issues.Append(fmt.Errorf("invalid user %d used for workflow run-as", wf.RunAs), nil) + } + } + + return } -func makeWorkflowHandler(ac workflowExecController, s *session, t *types.Trigger, wf *types.Workflow, g *wfexec.Graph, _runAs intAuth.Identifiable) eventbus.HandlerFn { +func (svc *workflow) exec(ctx context.Context, wf *types.Workflow, p types.WorkflowExecParams) (WaitFn, error) { + if wf.Issues != nil { + return nil, wf.Issues + } + + defer svc.mux.Unlock() + svc.mux.Lock() + + if svc.cache[wf.ID] == nil { + return nil, WorkflowErrInvalidID() + } + + var ( + g = svc.cache[wf.ID].g + runAs = svc.cache[wf.ID].runAs + + scope *expr.Vars + ) + + // merge workflow scope with the input + scope = wf.Scope.MustMerge(p.Input) + + // User (either invoker or one set in the security descriptor) MUST have + // permissions to execute this workflow + if !svc.ac.CanExecuteWorkflow(ctx, wf) { + return nil, WorkflowErrNotAllowedToExecute() + } + + return svc.session.Start(ctx, g, types.SessionStartParams{ + Invoker: intAuth.GetIdentityFromContext(ctx), + Runner: runAs, + + WorkflowID: wf.ID, + KeepFor: wf.KeepSessions, + Trace: wf.Trace, + Input: scope, + StepID: p.StepID, + EventType: p.EventType, + ResourceType: p.ResourceType, + + CallStack: wfexec.GetContextCallStack(ctx), + }) +} + +func makeWorkflowHandler(svc *workflow, wf *types.Workflow, t *types.Trigger) eventbus.HandlerFn { return func(ctx context.Context, ev eventbus.Event) (err error) { var ( - // create session scope from predefined workflow scope and trigger input - scope = wf.Scope.MustMerge(t.Input) - evScope *expr.Vars - wait WaitFn - - // The returned closure needs to have its own instance, so it doesn't - // affect the instance bound to the workflow handler - runAs = _runAs + scope *expr.Vars ) - if enc, is := ev.(varsEncoder); is { - if evScope, err = enc.EncodeVars(); err != nil { + if dec, is := ev.(varsEncoder); is { + scope, err = dec.EncodeVars() + if err != nil { return } - - scope = scope.MustMerge(evScope) } - _ = scope.AssignFieldValue("eventType", expr.Must(expr.NewString(ev.EventType()))) - _ = scope.AssignFieldValue("resourceType", expr.Must(expr.NewString(ev.ResourceType()))) - - if runAs == nil { - // @todo can/should we get alternative identity from Event? - // for example: - // - 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() - } - - callStack := wfexec.GetContextCallStack(ctx) - if len(callStack) > s.opt.CallStackSize { - return WorkflowErrMaximumCallStackSizeExceeded() - } - - wait, err = s.Start(g, runAs, types.SessionStartParams{ - WorkflowID: wf.ID, - KeepFor: wf.KeepSessions, - Trace: wf.Trace, - Input: scope, + wait, err := svc.exec(ctx, wf, types.WorkflowExecParams{ StepID: t.StepID, EventType: t.EventType, ResourceType: t.ResourceType, + Input: t.Input.MustMerge(scope), - CallStack: callStack, + Trace: wf.Trace, + Async: false, }) if err != nil { diff --git a/automation/service/workflow_converter.go b/automation/service/workflow_converter.go index e380c017a..5d24f54db 100644 --- a/automation/service/workflow_converter.go +++ b/automation/service/workflow_converter.go @@ -3,13 +3,14 @@ package service import ( "context" "fmt" + "strings" + "github.com/cortezaproject/corteza-server/automation/types" "github.com/cortezaproject/corteza-server/pkg/auth" "github.com/cortezaproject/corteza-server/pkg/errors" "github.com/cortezaproject/corteza-server/pkg/expr" "github.com/cortezaproject/corteza-server/pkg/wfexec" "go.uber.org/zap" - "strings" ) type ( @@ -18,6 +19,10 @@ type ( reg *registry parser expr.Parsable log *zap.Logger + + graphs interface { + handleToID(string) uint64 + } } ) @@ -26,6 +31,7 @@ func Convert(wfService *workflow, wf *types.Workflow) (*wfexec.Graph, types.Work reg: wfService.reg, parser: wfService.parser, log: wfService.log, + graphs: wfService, } return conv.makeGraph(wf) @@ -92,7 +98,7 @@ func (svc workflowConverter) makeGraph(def *types.Workflow) (*wfexec.Graph, type stepIssues := verifyStep(step, inPaths, outPaths) - if resolved, err := svc.workflowStepDefConv(g, step, inPaths, outPaths); err != nil { + if resolved, err := svc.workflowStepDefConv(g, def, step, inPaths, outPaths); err != nil { switch aux := err.(type) { case types.WorkflowIssueSet: stepIssues = append(stepIssues, aux...) @@ -151,7 +157,7 @@ func (svc workflowConverter) makeGraph(def *types.Workflow) (*wfexec.Graph, type // converts all step definitions into workflow.Step instances // // if this func returns nil for step and error, assume unresolved dependencies -func (svc workflowConverter) workflowStepDefConv(g *wfexec.Graph, s *types.WorkflowStep, in, out []*types.WorkflowPath) (bool, error) { +func (svc workflowConverter) workflowStepDefConv(g *wfexec.Graph, def *types.Workflow, s *types.WorkflowStep, in, out []*types.WorkflowPath) (bool, error) { if err := svc.parseExpressions(s.Arguments...); err != nil { return false, errors.Internal("failed to parse step arguments expressions for %s: %s", s.Kind, err).Wrap(err) } @@ -314,7 +320,7 @@ func (svc workflowConverter) convExpressionStep(s *types.WorkflowStep) (wfexec.S } // internal debug step that can log entire -func (svc workflowConverter) convDebugStep(s *types.WorkflowStep) (wfexec.Step, error) { +func (svc workflowConverter) convDebugStep(_ *types.WorkflowStep) (wfexec.Step, error) { return types.DebugStep(svc.log), nil } @@ -443,7 +449,7 @@ func (svc workflowConverter) convPromptStep(s *types.WorkflowStep) (wfexec.Step, } var ownerId uint64 = 0 - if i := auth.GetIdentityFromContext(ctx); i != nil { + if i := auth.GetIdentityFromContextWithKey(ctx, workflowInvokerCtxKey{}); i != nil { ownerId = i.Identity() } @@ -558,6 +564,16 @@ func verifyStep(s *types.WorkflowStep, in, out types.WorkflowPathSet) types.Work return nil } + // check if reference is set on the step + numericRef = func() error { + // @todo SUBWF enable this back + //if 0 == cast.ToUint64(s.Ref) { + // return errors.Internal("%s step expects workflow reference ID", s.Kind) + //} + + return nil + } + // reference should not be set on the step noRef = func() error { if s.Ref != "" { @@ -672,6 +688,7 @@ func verifyStep(s *types.WorkflowStep, in, out types.WorkflowPathSet) types.Work case types.WorkflowStepKindFunction: checks = append(checks, requiredRef, + numericRef, checkDisabledFunc, count(0, 1, outbound), ) diff --git a/automation/types/session.go b/automation/types/session.go index b2ee650a6..608bace8e 100644 --- a/automation/types/session.go +++ b/automation/types/session.go @@ -8,6 +8,7 @@ import ( "sync" "time" + "github.com/cortezaproject/corteza-server/pkg/auth" "github.com/cortezaproject/corteza-server/pkg/expr" "github.com/cortezaproject/corteza-server/pkg/filter" "github.com/cortezaproject/corteza-server/pkg/wfexec" @@ -53,6 +54,12 @@ type ( } SessionStartParams struct { + // Always set, users that invoked/started the workflow session + Invoker auth.Identifiable + + // Optional, (alternative) user that is running the workflow + Runner auth.Identifiable + WorkflowID uint64 KeepFor int Trace bool @@ -194,6 +201,42 @@ func (set Stacktrace) Value() (driver.Value, error) { return json.Marshal(set) } +func (set Stacktrace) String() (str string) { + for i, f := range set { + str += fmt.Sprintf( + "[%3d] %-14s %d (", + i, + f.CreatedAt.Format("15:04:05.00000"), + f.StepID, + ) + + if f.Input.Len() == 0 { + str += "no input, " + } + + if f.Scope.Len() == 0 { + str += "no scope, " + } + + if f.Results.Len() == 0 { + str += "no results, " + } + str += ")" + + if f.Scope.Len() > 0 { + str += fmt.Sprintf(" Scope:\n") + f.Scope.Each(func(k string, v expr.TypedValue) error { + str += fmt.Sprintf(" [%s]: %v\n", k, v) + return nil + }) + } + + str += "\n" + } + + return +} + func (s SessionStatus) String() string { switch s { case SessionStarted: diff --git a/automation/types/step.go b/automation/types/step.go index 3b9706d37..3c60d5efb 100644 --- a/automation/types/step.go +++ b/automation/types/step.go @@ -5,6 +5,7 @@ import ( "database/sql/driver" "encoding/json" "fmt" + "github.com/cortezaproject/corteza-server/pkg/expr" ) diff --git a/automation/types/workflow.go b/automation/types/workflow.go index 831a8029c..c86e4e39e 100644 --- a/automation/types/workflow.go +++ b/automation/types/workflow.go @@ -79,9 +79,21 @@ type ( } WorkflowExecParams struct { + // When executed as a sub-workflow + CallerWorkflowID uint64 + + // When executed as a sub-workflow + CallerSessionID uint64 + + // When executed as a sub-workflow + CallerStepID uint64 + // Start with this specific step StepID uint64 + EventType string + ResourceType string + // Enable execution tracing Trace bool @@ -105,6 +117,11 @@ func (r Workflow) CheckDeferred() bool { return r.Steps.HasDeferred() } +// Executable returns true if workflow is valid and enabled +func (r Workflow) Executable() bool { + return r.DeletedAt == nil && r.Enabled +} + func (r Workflow) Dict() map[string]interface{} { return map[string]interface{}{ "ID": r.ID, diff --git a/pkg/auth/identity_context.go b/pkg/auth/identity_context.go index 9d5f9f216..bde547521 100644 --- a/pkg/auth/identity_context.go +++ b/pkg/auth/identity_context.go @@ -16,9 +16,17 @@ func SetIdentityToContext(ctx context.Context, identity Identifiable) context.Co // // For anonymous user, it auto appends all anonymous defined on the system func GetIdentityFromContext(ctx context.Context) Identifiable { - if identity, ok := ctx.Value(identityCtxKey{}).(Identifiable); ok && identity != nil && identity.Valid() { - return identity + if i := GetIdentityFromContextWithKey(ctx, identityCtxKey{}); i != nil && i.Valid() { + return i } else { return Anonymous() } } + +func GetIdentityFromContextWithKey(ctx context.Context, key interface{}) Identifiable { + if i, ok := ctx.Value(key).(Identifiable); ok { + return i + } else { + return nil + } +}