marker/shuffle-cracked
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixed most of user input continuations

Browse Source
This commit is contained in:
frikky
2020-11-16 18:37:04 +01:00
parent 3f98f986e7
commit ae39da8ffd
6 changed files with 400 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/go-app/main.go
+106 -25
View File
@@ -74,9 +74,10 @@ var baseAppPath = "/home/frikky/git/shaffuru/tmp/apps"
var baseDockerName = "frikky/shuffle"
//var syncUrl = "http://192.168.102.54:5002"
//var syncUrl = "http://localhost:5002"
var syncUrl = "https://shuffler.io"
//var syncUrl = "http://localhost:5002"
var dbclient *datastore.Client
type Userapi struct {
@@ -3682,9 +3683,10 @@ func handleNewHook(resp http.ResponseWriter, request *http.Request) {
// Let remote endpoint handle access checks (shuffler.io)
currentUrl := fmt.Sprintf("https://shuffler.io/api/v1/hooks/webhook_%s", newId)
startNode := requestdata.Start
if requestdata.Environment == "cloud" {
log.Printf("[INFO] Should START a cloud webhook for url %s", currentUrl)
// https://shuffler.io/v1/hooks/webhook_80184973-3e82-4852-842e-0290f7f34d7c
log.Printf("[INFO] Should START a cloud webhook for url %s for startnode %s", currentUrl, startNode)
org, err := getOrg(ctx, user.ActiveOrg.Id)
if err != nil {
log.Printf("Failed finding org %s: %s", org.Id, err)
@@ -3696,7 +3698,7 @@ func handleNewHook(resp http.ResponseWriter, request *http.Request) {
Action: "start",
OrgId: org.Id,
PrimaryItemId: newId,
SecondaryItem: requestdata.Start,
SecondaryItem: startNode,
ThirdItem: requestdata.Workflow,
}
@@ -3713,7 +3715,7 @@ func handleNewHook(resp http.ResponseWriter, request *http.Request) {
hook := Hook{
Id: newId,
Start: requestdata.Start,
Start: startNode,
Workflows: []string{requestdata.Workflow},
Info: Info{
Name: requestdata.Name,
@@ -4709,7 +4711,7 @@ Please contact us at shuffler.io or frikky@shuffler.io if there is an issue with
}
resp.WriteHeader(200)
resp.Write([]byte("OK"))
resp.Write([]byte(`{"success": true}`))
}
func setBadMemcache(ctx context.Context, path string) {
@@ -5186,7 +5188,7 @@ func handleNewOutlookRegister(resp http.ResponseWriter, request *http.Request) {
}
resp.WriteHeader(200)
resp.Write([]byte("OK"))
resp.Write([]byte(`{"success": true}`))
}
type OauthToken struct {
@@ -6657,22 +6659,34 @@ func handleCloudExecutionOnprem(workflowId, startNode, executionSource, executio
if err != nil {
return err
}
// FIXME: Handle auth
_ = workflow
type execStruct struct {
ExecutionSource string `json:"execution_source"`
ExecutionArgument string `json:"execution_argument"`
Start string `json:"start,omitempty"`
}
parsedArgument := strings.Replace(string(executionArgument), "\"", "\\\"", -1)
newExec := execStruct{
parsedArgument := executionArgument
newExec := ExecutionRequest{
ExecutionSource: executionSource,
ExecutionArgument: parsedArgument,
}
//bodyWrapper := fmt.Sprintf(`{"execution_source": "%s", "execution_argument": "%s"}`, executionSource, parsedArgument)
var execution ExecutionRequest
err = json.Unmarshal([]byte(parsedArgument), &execution)
if err == nil {
log.Printf("FOUND EXEC %#v", execution)
if len(execution.ExecutionArgument) > 0 {
parsedArgument := strings.Replace(string(execution.ExecutionArgument), "\\\"", "\"", -1)
log.Printf("New exec argument: %s", execution.ExecutionArgument)
if strings.HasPrefix(parsedArgument, "{") && strings.HasSuffix(parsedArgument, "}") {
log.Printf("\nData is most likely JSON from %s\n", newExec.ExecutionSource)
}
newExec.ExecutionArgument = parsedArgument
}
} else {
log.Printf("Unmarshal issue: %s", err)
}
if len(startNode) > 0 {
newExec.Start = startNode
}
@@ -6694,10 +6708,12 @@ func handleCloudExecutionOnprem(workflowId, startNode, executionSource, executio
}
func handleCloudJob(job CloudSyncJob) error {
// May need authentication in all of these..?
log.Printf("Handle job with type %s and action %s", job.Type, job.Action)
if job.Type == "webhook" {
if job.Action == "execute" {
log.Printf("Should handle webhook for workflow %s with start node %s and data %s", job.PrimaryItemId, job.SecondaryItem)
log.Printf("Should handle webhook for workflow %s with start node %s and data %s", job.PrimaryItemId, job.SecondaryItem, job.ThirdItem)
err := handleCloudExecutionOnprem(job.PrimaryItemId, job.SecondaryItem, "webhook", job.ThirdItem)
if err != nil {
log.Printf("Failed executing workflow from cloud hook: %s", err)
@@ -6728,14 +6744,79 @@ func handleCloudJob(job CloudSyncJob) error {
}
} else if job.Type == "user_input" {
if job.Action == "execute" {
log.Printf("Should handle user_input for workflow %s with start node %s and data %s", job.PrimaryItemId, job.SecondaryItem, job.ThirdItem)
err := handleCloudExecutionOnprem(job.PrimaryItemId, job.SecondaryItem, "user_input", job.ThirdItem)
if job.Action == "continue" {
log.Printf("Should handle user_input CONTINUE for workflow %s with start node %s and execution ID %s", job.PrimaryItemId, job.SecondaryItem, job.ThirdItem)
// FIXME: Handle authorization
ctx := context.Background()
workflowExecution, err := getWorkflowExecution(ctx, job.ThirdItem)
if err != nil {
log.Printf("Failed executing workflow from cloud user_input: %s", err)
return err
}
if job.PrimaryItemId != workflowExecution.Workflow.ID {
return errors.New("Bad workflow ID when stopping execution.")
}
workflowExecution.Status = "EXECUTING"
err = setWorkflowExecution(ctx, *workflowExecution)
if err != nil {
return err
}
fullUrl := fmt.Sprintf("https://shuffler.io/api/v1/workflows/%s/execute?authorization=%s&start=%s&reference_execution=%s&answer=true", job.PrimaryItemId, job.FourthItem, job.SecondaryItem, job.ThirdItem)
newRequest, err := http.NewRequest(
"GET",
fullUrl,
nil,
)
if err != nil {
log.Printf("Failed continuing workflow in request builder: %s", err)
return err
}
_, _, err = handleExecution(job.PrimaryItemId, Workflow{}, newRequest)
if err != nil {
log.Printf("Failed continuing workflow from cloud user_input: %s", err)
return err
} else {
log.Printf("Successfully executed workflow from cloud user_input")
}
} else if job.Action == "stop" {
log.Printf("Should handle user_input STOP for workflow %s with start node %s and execution ID %s", job.PrimaryItemId, job.SecondaryItem, job.ThirdItem)
ctx := context.Background()
workflowExecution, err := getWorkflowExecution(ctx, job.ThirdItem)
if err != nil {
return err
}
if job.PrimaryItemId != workflowExecution.Workflow.ID {
return errors.New("Bad workflow ID when stopping execution.")
}
/*
if job.FourthItem != workflowExecution.Authorization {
return errors.New("Bad authorization when stopping execution.")
}
*/
newResults := []ActionResult{}
for _, result := range workflowExecution.Results {
if result.Action.AppName == "User Input" && result.Result == "Waiting for user feedback based on configuration" {
result.Status = "ABORTED"
result.Result = "Aborted manually by user."
}
newResults = append(newResults, result)
}
workflowExecution.Results = newResults
workflowExecution.Status = "ABORTED"
err = setWorkflowExecution(ctx, *workflowExecution)
if err != nil {
return err
}
log.Printf("Successfully updated user input to aborted.")
}
} else {
log.Printf("No handler for type %s and action %s", job.Type, job.Action)
@@ -6763,7 +6844,7 @@ func remoteOrgJobController(org Org, body []byte) error {
log.Printf("Should stop org job controller")
if strings.Contains(responseData.Reason, "Bad apikey") {
log.Printf("Bad apikey. Stopping sync for org!")
log.Printf("Bad apikey. Stopping sync for org?: %s", responseData.Reason)
if value, exists := scheduledOrgs[org.Id]; exists {
// Looks like this does the trick? Hurr
@@ -6786,7 +6867,7 @@ func remoteOrgJobController(org Org, body []byte) error {
log.Printf("Successfully updated the org to not sync")
}
return errors.New("Stopped schedule for org because of bad apikey.")
return errors.New("Stopped schedule for org locally because of bad apikey.")
} else {
return errors.New(fmt.Sprintf("Failed finding the schedule for org %s", org.Id))
}
@@ -7223,7 +7304,7 @@ func runInit(ctx context.Context) {
}
//interval := int(org.SyncConfig.Interval)
interval := 5
interval := 15
if interval == 0 {
log.Printf("Skipping org %s because sync isn't set (0).", org.Id)
continue
@@ -7731,7 +7812,7 @@ func handleCloudSetup(resp http.ResponseWriter, request *http.Request) {
// return
//}
log.Printf("Apidata: %s", tmpData.Apikey)
//log.Printf("Apidata: %s", tmpData.Apikey)
// FIXME: Path
client := &http.Client{}
@@ -7808,7 +7889,7 @@ func handleCloudSetup(resp http.ResponseWriter, request *http.Request) {
return
}
log.Printf("Respbody: %s", string(respBody))
//log.Printf("Respbody: %s", string(respBody))
responseData := retStruct{}
err = json.Unmarshal(respBody, &responseData)
if err != nil {
backend/go-app/walkoff.go
+159 -79
View File
@@ -56,15 +56,15 @@ var scheduledOrgs = map[string]*newscheduler.Job{}
//}
type ExecutionRequest struct {
ExecutionId string `json:"execution_id"`
ExecutionArgument string `json:"execution_argument"`
ExecutionSource string `json:"execution_source"`
WorkflowId string `json:"workflow_id"`
Environments []string `json:"environments"`
Authorization string `json:"authorization"`
Status string `json:"status"`
Start string `json:"start"`
Type string `json:"type"`
ExecutionId string `json:"execution_id,omitempty"`
ExecutionArgument string `json:"execution_argument,omitempty"`
ExecutionSource string `json:"execution_source,omitempty"`
WorkflowId string `json:"workflow_id,omitempty"`
Environments []string `json:"environments,omitempty"`
Authorization string `json:"authorization,omitempty"`
Status string `json:"status,omitempty"`
Start string `json:"start,omitempty"`
Type string `json:"type,omitempty"`
}
type SyncFeatures struct {
@@ -685,7 +685,7 @@ func handleGetWorkflowqueueConfirm(resp http.ResponseWriter, request *http.Reque
//}
resp.WriteHeader(200)
resp.Write([]byte("OK"))
resp.Write([]byte(`{"success": true}`))
}
// FIXME: Authenticate this one? Can org ID be auth enough?
@@ -899,6 +899,48 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
}
}
if actionResult.Status == "WAITING" && actionResult.Action.AppName == "User Input" {
log.Printf("SHOULD WAIT A BIT AND RUN CLOUD STUFF WITH USER INPUT! WAITING!")
var trigger Trigger
err = json.Unmarshal([]byte(actionResult.Result), &trigger)
if err != nil {
log.Printf("Failed unmarshaling actionresult for user input: %s", err)
resp.WriteHeader(401)
resp.Write([]byte(`{"success": false}`))
return
}
orgId := workflowExecution.ExecutionOrg
if len(workflowExecution.OrgId) == 0 && len(workflowExecution.Workflow.OrgId) > 0 {
orgId = workflowExecution.Workflow.OrgId
}
err := handleUserInput(trigger, orgId, workflowExecution.Workflow.ID, workflowExecution.ExecutionId)
if err != nil {
log.Printf("Failed userinput handler: %s", err)
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Error: %s"}`, err)))
} else {
log.Printf("Successful userinput handler")
resp.WriteHeader(200)
resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "CLOUD IS DONE"}`)))
actionResult.Result = "Waiting for user feedback based on configuration"
workflowExecution.Results = append(workflowExecution.Results, actionResult)
workflowExecution.Status = actionResult.Status
err = setWorkflowExecution(ctx, *workflowExecution)
if err != nil {
log.Printf("Failed ")
} else {
log.Printf("Successfully set the execution to waiting.")
}
}
return
}
if actionResult.Status == "ABORTED" || actionResult.Status == "FAILURE" {
log.Printf("Actionresult is %s. Should set workflowExecution and exit all running functions", actionResult.Status)
@@ -977,7 +1019,7 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
for _, result := range workflowExecution.Results {
if result.Status == "EXECUTING" {
result.Status = actionResult.Status
result.Result = "Aborted because of an unknown error"
result.Result = "Aborted because of error in another node"
}
if len(result.Result) > 0 {
@@ -1201,10 +1243,10 @@ func handleExecutionStatistics(execution WorkflowExecution) {
for _, result := range execution.Results {
resultCheck := JSONCheck(result.Result)
if !resultCheck {
log.Printf("Result is NOT JSON!")
//log.Printf("Result is NOT JSON!")
continue
} else {
log.Printf("Result IS JSON!")
//log.Printf("Result IS JSON!")
}
@@ -1839,6 +1881,13 @@ func saveWorkflow(resp http.ResponseWriter, request *http.Request) {
}
}
if len(triggerType) == 0 {
log.Printf("No type specified for user input node")
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "No contact option specified in user input"}`)))
return
}
// FIXME: This is not the right time to send them, BUT it's well served for testing. Save -> send email / sms
_ = triggerInformation
if strings.Contains(triggerType, "email") {
@@ -1849,38 +1898,6 @@ func saveWorkflow(resp http.ResponseWriter, request *http.Request) {
return
}
// triggerid, start node, workflowid, argument
/*
startNode := ""
referenceExecutionId := ""
action := CloudSyncJob{
Type: "user_input",
Action: "send_email",
OrgId: user.ActiveOrg.Id,
PrimaryItemId: workflow.ID,
SecondaryItem: startNode,
ThirdItem: triggerInformation,
FourthItem: email,
FifthItem: referenceExecutionId,
}
org, err := getOrg(ctx, user.ActiveOrg.Id)
if err != nil {
log.Printf("Failed email send to cloud", err)
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err)))
return
}
err = executeCloudAction(action, org.SyncConfig.Apikey)
if err != nil {
log.Printf("Failed email send to cloud", err)
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err)))
return
}
*/
log.Printf("Should send email to %s during execution.", email)
}
if strings.Contains(triggerType, "sms") {
@@ -1891,37 +1908,6 @@ func saveWorkflow(resp http.ResponseWriter, request *http.Request) {
return
}
/*
startNode := ""
referenceExecutionId := ""
action := CloudSyncJob{
Type: "user_input",
Action: "send_sms",
OrgId: user.ActiveOrg.Id,
PrimaryItemId: workflow.ID,
SecondaryItem: startNode,
ThirdItem: triggerInformation,
FourthItem: sms,
FifthItem: referenceExecutionId,
}
org, err := getOrg(ctx, user.ActiveOrg.Id)
if err != nil {
log.Printf("Failed email send to cloud", err)
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err)))
return
}
err = executeCloudAction(action, org.SyncConfig.Apikey)
if err != nil {
log.Printf("Failed email send to cloud", err)
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err)))
return
}
*/
log.Printf("Should send SMS to %s during execution.", sms)
}
}
@@ -2346,7 +2332,7 @@ func abortExecution(resp http.ResponseWriter, request *http.Request) {
return
}
} else {
log.Printf("API key %s is correct to abort %s", parsedKey, executionId)
log.Printf("API key to abort/finish execution %s is correct.", executionId)
}
if workflowExecution.Status == "ABORTED" || workflowExecution.Status == "FAILURE" || workflowExecution.Status == "FINISHED" {
@@ -2367,7 +2353,7 @@ func abortExecution(resp http.ResponseWriter, request *http.Request) {
for _, result := range workflowExecution.Results {
if result.Status == "EXECUTING" {
result.Status = "ABORTED"
result.Result = "Aborted because of an unknown error"
result.Result = "Aborted because of error in another node"
}
if len(result.Result) > 0 {
@@ -2446,7 +2432,7 @@ func cleanupExecutions(resp http.ResponseWriter, request *http.Request) {
}
resp.WriteHeader(200)
resp.Write([]byte("OK"))
resp.Write([]byte(`{"success": true}`))
}
func handleExecution(id string, workflow Workflow, request *http.Request) (WorkflowExecution, string, error) {
@@ -2507,6 +2493,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
}
// This one doesn't really matter.
log.Printf("Running POST execution with data %s", body)
var execution ExecutionRequest
err = json.Unmarshal(body, &execution)
if err != nil {
@@ -2797,6 +2784,16 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
}
}
for _, trigger := range workflowExecution.Workflow.Triggers {
log.Printf("ID: %s vs %s", trigger.ID, workflowExecution.Start)
if trigger.ID == workflowExecution.Start {
if trigger.AppName == "User Input" {
startFound = true
break
}
}
}
if !startFound {
log.Printf("Startnode %s doesn't exist!", workflowExecution.Start)
return WorkflowExecution{}, fmt.Sprintf("Workflow action %s doesn't exist in workflow", workflowExecution.Start), errors.New(fmt.Sprintf(`Workflow start node "%s" doesn't exist. Exiting!`, workflowExecution.Start))
@@ -3509,6 +3506,7 @@ func scheduleWorkflow(resp http.ResponseWriter, request *http.Request) {
SecondaryItem: schedule.Frequency,
ThirdItem: workflow.ID,
FourthItem: schedule.ExecutionArgument,
FifthItem: startNode,
}
timeNow := int64(time.Now().Unix())
@@ -6128,3 +6126,85 @@ func removeOutlookTriggerFunction(ctx context.Context, triggerId string) error {
_ = resp
return nil
}
func handleUserInput(trigger Trigger, organizationId string, workflowId string, referenceExecution string) error {
// E.g. check email
sms := ""
email := ""
triggerType := ""
triggerInformation := ""
for _, item := range trigger.Parameters {
if item.Name == "alertinfo" {
triggerInformation = item.Value
} else if item.Name == "type" {
triggerType = item.Value
} else if item.Name == "email" {
email = item.Value
} else if item.Name == "sms" {
sms = item.Value
}
}
if len(triggerType) == 0 {
log.Printf("No type specified for user input node")
return errors.New("No type specified for user input node")
}
// FIXME: This is not the right time to send them, BUT it's well served for testing. Save -> send email / sms
ctx := context.Background()
startNode := trigger.ID
if strings.Contains(triggerType, "email") {
action := CloudSyncJob{
Type: "user_input",
Action: "send_email",
OrgId: organizationId,
PrimaryItemId: workflowId,
SecondaryItem: startNode,
ThirdItem: triggerInformation,
FourthItem: email,
FifthItem: referenceExecution,
}
org, err := getOrg(ctx, organizationId)
if err != nil {
log.Printf("Failed email send to cloud: %s", err)
return err
}
err = executeCloudAction(action, org.SyncConfig.Apikey)
if err != nil {
log.Printf("Failed email send to cloud", err)
return err
}
log.Printf("Should send email to %s during execution.", email)
}
if strings.Contains(triggerType, "sms") {
action := CloudSyncJob{
Type: "user_input",
Action: "send_sms",
OrgId: organizationId,
PrimaryItemId: workflowId,
SecondaryItem: startNode,
ThirdItem: triggerInformation,
FourthItem: sms,
FifthItem: referenceExecution,
}
org, err := getOrg(ctx, organizationId)
if err != nil {
log.Printf("Failed email send to cloud", err)
return err
}
err = executeCloudAction(action, org.SyncConfig.Apikey)
if err != nil {
log.Printf("Failed email send to cloud", err)
return err
}
log.Printf("Should send SMS to %s during execution.", sms)
}
return nil
}