Merge branch '2.0.0'

This commit is contained in:
Aditya
2024-12-05 05:20:50 +05:30
136 changed files with 45316 additions and 19511 deletions
+16 -9
View File
@@ -211,6 +211,7 @@ func fixTags(tags []string) []string {
func buildImageMemory(fs billy.Filesystem, tags []string, dockerfileFolder string, downloadIfFail bool) error {
ctx := context.Background()
client, err := client.NewEnvClient()
defer client.Close()
if err != nil {
log.Printf("Unable to create docker client: %s", err)
return err
@@ -349,7 +350,7 @@ func deleteJob(client *kubernetes.Clientset, jobName, namespace string) error {
})
}
func buildImage(tags []string, dockerfileFolder string) error {
func buildImage(tags []string, dockerfileLocation string) error {
isKubernetes := false
if os.Getenv("IS_KUBERNETES") == "true" {
@@ -369,10 +370,8 @@ func buildImage(tags []string, dockerfileFolder string) error {
log.Printf("[INFO] registry name: %s", registryName)
contextDir := strings.Replace(dockerfileFolder, "Dockerfile", "", -1)
contextDir = "/app/" + contextDir
contextDir := filepath.Join("/app/", filepath.Dir(dockerfileLocation))
log.Print("contextDir: ", contextDir)
dockerFile := "./Dockerfile"
client, err := getK8sClient()
if err != nil {
@@ -407,7 +406,7 @@ func buildImage(tags []string, dockerfileFolder string) error {
Image: "gcr.io/kaniko-project/executor:latest",
Args: []string{
"--verbosity=debug",
"--dockerfile=" + dockerFile,
"--dockerfile=Dockerfile",
"--context=dir://" + contextDir,
"--skip-tls-verify",
"--destination=" + registryName + "/" + tags[1],
@@ -420,9 +419,7 @@ func buildImage(tags []string, dockerfileFolder string) error {
},
},
},
NodeSelector: map[string]string{
"node": backendNodeName,
},
NodeName: backendNodeName,
RestartPolicy: corev1.RestartPolicyNever,
Volumes: []corev1.Volume{
{
@@ -480,13 +477,14 @@ func buildImage(tags []string, dockerfileFolder string) error {
ctx := context.Background()
client, err := client.NewEnvClient()
defer client.Close()
if err != nil {
log.Printf("Unable to create docker client: %s", err)
return err
}
log.Printf("[INFO] Docker Tags: %s", tags)
dockerfileSplit := strings.Split(dockerfileFolder, "/")
dockerfileSplit := strings.Split(dockerfileLocation, "/")
// Create a buffer
buf := new(bytes.Buffer)
@@ -836,14 +834,23 @@ func handleRemoteDownloadApp(resp http.ResponseWriter, ctx context.Context, user
type tmpapp struct {
Success bool `json:"success"`
OpenAPI string `json:"openapi"`
App string `json:"app"`
}
app := tmpapp{}
err := json.Unmarshal(respBody, &app)
if err != nil || app.Success == false || len(app.OpenAPI) == 0 {
log.Printf("[ERROR] Failed app unmarshal during auto-download. Success: %#v. Applength: %d: %s", app.Success, len(app.OpenAPI), err)
resp.WriteHeader(401)
if len(app.App) > 0 {
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Not an OpenAPI app, but a Python app. Please download the app using the Remote Download system: https://shuffler.io/docs/apps#importing-remote-apps"}`)))
} else {
resp.Write([]byte(`{"success": false, "reason": "App doesn't exist"}`))
}
resp.Write([]byte(`{"success": false, "reason": "App doesn't exist"}`))
return
}
+2 -2
View File
@@ -2,7 +2,7 @@ module shuffle
go 1.22.0
// replace github.com/shuffle/shuffle-shared => ../../../shuffle-shared
//replace github.com/shuffle/shuffle-shared => ../../../shuffle-shared
toolchain go1.22.2
@@ -20,7 +20,7 @@ require (
github.com/gorilla/mux v1.8.1
github.com/h2non/filetype v1.1.3
github.com/satori/go.uuid v1.2.0
github.com/shuffle/shuffle-shared v0.6.50
github.com/shuffle/shuffle-shared v0.6.90
golang.org/x/crypto v0.22.0
google.golang.org/api v0.176.1
google.golang.org/grpc v1.63.2
+4 -2
View File
@@ -334,8 +334,10 @@ github.com/sendgrid/sendgrid-go v3.14.0+incompatible/go.mod h1:QRQt+LX/NmgVEvmdR
github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3 h1:n661drycOFuPLCN3Uc8sB6B/s6Z4t2xvBgU1htSHuq8=
github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3/go.mod h1:A0bzQcvG0E7Rwjx0REVgAGH58e96+X0MeOfepqsbeW4=
github.com/shuffle/shuffle-shared v0.6.50 h1:MBeGAiBNkw9Eg+3YTJIlBOuskWntGvT0uefFUYOBhbY=
github.com/shuffle/shuffle-shared v0.6.50/go.mod h1:RAJiSFjmuKmijKTbbEf9A6Ojb+3/te7g71lED7JjPus=
github.com/shuffle/shuffle-shared v0.6.77 h1:KKtM50xW2DLuRHINxhp3uXrNH0AhiwkeiiU93a8fB3A=
github.com/shuffle/shuffle-shared v0.6.77/go.mod h1:RAJiSFjmuKmijKTbbEf9A6Ojb+3/te7g71lED7JjPus=
github.com/shuffle/shuffle-shared v0.6.90 h1:FzIYtEt44eWgEsW/9tj2ki7qq8FEm/HWXUok+THp72M=
github.com/shuffle/shuffle-shared v0.6.90/go.mod h1:RAJiSFjmuKmijKTbbEf9A6Ojb+3/te7g71lED7JjPus=
github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
+202 -71
View File
@@ -35,6 +35,7 @@ import (
"github.com/go-git/go-billy/v5/memfs"
"github.com/go-git/go-git/v5"
"github.com/go-git/go-git/v5/plumbing"
gitProxy "github.com/go-git/go-git/v5/plumbing/transport"
"github.com/go-git/go-git/v5/storage/memory"
// Random
@@ -256,7 +257,6 @@ type Hook struct {
Environment string `json:"environment" datastore:"environment"`
}
func GetUsersHandler(w http.ResponseWriter, r *http.Request) {
data := map[string]interface{}{
"id": "12345",
@@ -396,6 +396,52 @@ func checkUsername(Username string) error {
return nil
}
func isGitNoProxy(rawURL string) bool {
noProxy := os.Getenv("NO_PROXY")
if noProxy == "" {
return false
}
if noProxy == "*" {
return true
}
noProxyList := strings.Split(noProxy, ",")
parsedURL, err := url.Parse(rawURL)
if err != nil {
return false
}
host := parsedURL.Hostname()
for _, value := range noProxyList {
value = strings.TrimSpace(value)
if host == value {
return true
}
if strings.HasPrefix(value, "*.") && strings.HasSuffix(host, value[2:]) {
return true
}
}
return false
}
func checkGitProxy(cloneOptions *git.CloneOptions) *git.CloneOptions {
if os.Getenv("HTTP_PROXY") != "" && !isGitNoProxy(cloneOptions.URL) {
cloneOptions.ProxyOptions = gitProxy.ProxyOptions{
URL: os.Getenv("HTTP_PROXY"),
}
}
if os.Getenv("HTTPS_PROXY") != "" && !isGitNoProxy(cloneOptions.URL) {
cloneOptions.ProxyOptions = gitProxy.ProxyOptions{
URL: os.Getenv("HTTPS_PROXY"),
}
}
return cloneOptions
}
func createNewUser(username, password, role, apikey string, org shuffle.OrgMini) error {
// Returns false if there is an issue
// Use this for register
@@ -450,6 +496,7 @@ func createNewUser(username, password, role, apikey string, org shuffle.OrgMini)
newUser.ActiveOrg = shuffle.OrgMini{
Id: org.Id,
Name: org.Name,
Role: newUser.Role,
}
if len(apikey) > 0 {
@@ -511,7 +558,6 @@ func createNewUser(username, password, role, apikey string, org shuffle.OrgMini)
}
}
return nil
}
@@ -615,7 +661,7 @@ func handleRegister(resp http.ResponseWriter, request *http.Request) {
Name: newOrg.Name,
}
user.ActiveOrg = currentOrg
user.ActiveOrg = currentOrg
}
}
}
@@ -884,18 +930,58 @@ func handleInfo(resp http.ResponseWriter, request *http.Request) {
log.Printf("[DEBUG] Failed to get org during getinfo: %s", err)
}
//if err == nil {
if len(org.Id) > 0 {
if userInfo.Role == "" {
//err = shuffle.SetUser(ctx, &userInfo, false)
for _, user := range org.Users {
if user.Id != userInfo.Id {
continue
}
userInfo.ActiveOrg.Role = user.Role
}
}
userInfo.ActiveOrg = shuffle.OrgMini{
Id: org.Id,
Name: org.Name,
CreatorOrg: org.CreatorOrg,
ChildOrgs: org.ChildOrgs,
Role: userInfo.ActiveOrg.Role,
Image: org.Image,
}
if parsedAdmin == "false" {
// Validating admin user again just to make sure
// This is to avoid issues for the first org ever
for _, user := range org.Users {
if user.Id != userInfo.Id {
continue
}
if user.Role == "admin" {
break
}
}
}
}
//}
orgPriorities := org.Priorities
if len(org.Priorities) < 10 {
//log.Printf("[WARNING] Should find and add priorities as length is less than 10 for org %s", userInfo.ActiveOrg.Id)
newPriorities, err := shuffle.GetPriorities(ctx, userInfo, org)
if err != nil {
log.Printf("[WARNING] Failed getting new priorities for org %s: %s", org.Id, err)
//orgPriorities = []shuffle.Priority{}
} else {
orgPriorities = newPriorities
// A way to manage them over time
}
}
orgInterests := org.Interests
userInfo.ActiveOrg.Users = []shuffle.UserMini{}
userOrgs := []shuffle.OrgMini{}
@@ -942,19 +1028,6 @@ func handleInfo(resp http.ResponseWriter, request *http.Request) {
}
userOrgs = shuffle.SortOrgList(userOrgs)
orgPriorities := org.Priorities
if len(org.Priorities) < 10 {
//log.Printf("[WARNING] Should find and add priorities as length is less than 10 for org %s", userInfo.ActiveOrg.Id)
newPriorities, err := shuffle.GetPriorities(ctx, userInfo, org)
if err != nil {
log.Printf("[WARNING] Failed getting new priorities for org %s: %s", org.Id, err)
//orgPriorities = []shuffle.Priority{}
} else {
orgPriorities = newPriorities
// A way to manage them over time
}
}
tutorialsFinished := []shuffle.Tutorial{}
for _, tutorial := range userInfo.PersonalInfo.Tutorials {
@@ -993,8 +1066,9 @@ func handleInfo(resp http.ResponseWriter, request *http.Request) {
ChatDisabled: chatDisabled,
Tutorials: tutorialsFinished,
Interests: orgInterests,
Priorities: orgPriorities,
Licensed: licensed,
Licensed: licensed,
}
returnData, err := json.Marshal(returnValue)
@@ -1015,7 +1089,6 @@ type passwordReset struct {
Reference string `json:"reference"`
}
func checkAdminLogin(resp http.ResponseWriter, request *http.Request) {
cors := shuffle.HandleCors(resp, request)
if cors {
@@ -1054,9 +1127,9 @@ func checkAdminLogin(resp http.ResponseWriter, request *http.Request) {
}
// No childorg setup, only parent org
if len(org.ManagerOrgs) > 0 || len(org.CreatorOrg) > 0 {
continue
}
// if len(org.ManagerOrgs) > 0 || len(org.CreatorOrg) > 0 {
// continue
// }
// Should run calculations
if len(org.SSOConfig.OpenIdAuthorization) > 0 {
@@ -1978,7 +2051,6 @@ func handleWebhookCallback(resp http.ResponseWriter, request *http.Request) {
}
func handlePipelineCallback(resp http.ResponseWriter, request *http.Request) {
if request.Method != "POST" {
request.Method = "POST"
}
@@ -1999,7 +2071,7 @@ func handlePipelineCallback(resp http.ResponseWriter, request *http.Request) {
location := strings.Split(request.URL.String(), "/")
var pipelineId string
if location[1] == "api" {
if len(location) <= 4 {
log.Printf("[INFO] Couldn't handle location. Too short in pipeline: %d", len(location))
@@ -2013,7 +2085,7 @@ func handlePipelineCallback(resp http.ResponseWriter, request *http.Request) {
userAgent := request.Header.Get("User-Agent")
if strings.Contains(strings.ToLower(userAgent), "microsoftpreview") || strings.Contains(strings.ToLower(userAgent), "googlebot") {
log.Printf("[AUDIT] Blocking googlebot and microsoftbot for pielines. UA: '%s'", userAgent)
log.Printf("[AUDIT] Blocking googlebot and microsoftbot for pipelines. UA: '%s'", userAgent)
resp.WriteHeader(400)
resp.Write([]byte(`{"success": false, "reason": "Google/Microsoft preview bots not allowed. Please change the useragent."}`))
return
@@ -2058,11 +2130,27 @@ func handlePipelineCallback(resp http.ResponseWriter, request *http.Request) {
return
}
parsedBody := shuffle.GetExecutionbody(body)
// Parse concatenated JSON logs
jsonList, err := parseConcatenatedJSONLogs(string(body))
if err != nil {
log.Printf("[DEBUG] JSON parsing error: %s", err)
resp.WriteHeader(401)
resp.Write([]byte(`{"success": false}`))
return
}
parsedBody, err := json.Marshal(jsonList)
if err != nil {
log.Printf("[ERROR] Failed to marshal jsonList: %s", err)
resp.WriteHeader(500)
resp.Write([]byte(`{"success": false}`))
return
}
newBody := shuffle.ExecutionStruct{
Start: pipeline.StartNode,
ExecutionSource: "pipeline",
ExecutionArgument: parsedBody,
ExecutionArgument: string(parsedBody),
}
workflow, err := shuffle.GetWorkflow(ctx, pipeline.WorkflowId)
@@ -2093,8 +2181,7 @@ func handlePipelineCallback(resp http.ResponseWriter, request *http.Request) {
}
if len(pipeline.StartNode) == 0 {
log.Printf("[WARNING] No start node for pipeline %s - running with workflow default.", pipeline.TriggerId)
log.Printf("[WARNING] No start node for pipeline %s - running with workflow default.")
}
newRequest := &http.Request{
@@ -2108,6 +2195,9 @@ func handlePipelineCallback(resp http.ResponseWriter, request *http.Request) {
if err == nil {
resp.WriteHeader(200)
resp.Write([]byte(fmt.Sprintf(`{"success": true, "execution_id": "%s"}`, workflowExecution.ExecutionId)))
// Track Sigma rules
trackSigmaRules(ctx, pipeline.OrgId, jsonList)
return
}
@@ -2115,6 +2205,42 @@ func handlePipelineCallback(resp http.ResponseWriter, request *http.Request) {
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, executionResp)))
}
func parseConcatenatedJSONLogs(logs string) ([]map[string]interface{}, error) {
var jsonList []map[string]interface{}
decoder := json.NewDecoder(strings.NewReader(logs))
for decoder.More() {
var jsonObject map[string]interface{}
if err := decoder.Decode(&jsonObject); err != nil {
log.Printf("[WARNING] JSON decoding error: %s. Skipping this object.", err)
continue
}
jsonList = append(jsonList, jsonObject)
}
if err := decoder.Decode(&struct{}{}); err != io.EOF {
return nil, fmt.Errorf("error after decoding all JSON objects: %v", err)
}
return jsonList, nil
}
func trackSigmaRules(ctx context.Context, orgId string, jsonList []map[string]interface{}) {
ruleCount := make(map[string]int)
for _, logEntry := range jsonList {
if rule, ok := logEntry["rule"].(map[string]interface{}); ok {
if ruleName, ok := rule["title"].(string); ok {
ruleCount[ruleName]++
}
}
}
for ruleName, count := range ruleCount {
shuffle.IncrementCache(ctx, orgId, ruleName, count)
log.Printf("[INFO] Rule %s incremented by %d", ruleName, count)
}
}
func executeCloudAction(action shuffle.CloudSyncJob, apikey string) error {
data, err := json.Marshal(action)
if err != nil {
@@ -3208,7 +3334,6 @@ func buildSwaggerApp(resp http.ResponseWriter, body []byte, user shuffle.User, s
}
}
log.Printf("[DEBUG] Successfully built app %s (%s)", api.Name, api.ID)
if len(user.Id) > 0 {
resp.WriteHeader(200)
@@ -3249,8 +3374,6 @@ func verifySwagger(resp http.ResponseWriter, request *http.Request) {
buildSwaggerApp(resp, body, user, false)
}
// Hotloads new apps from a folder
func handleAppHotload(ctx context.Context, location string, forceUpdate bool) error {
@@ -3597,11 +3720,10 @@ func remoteOrgJobController(org shuffle.Org, body []byte) error {
return nil
}
func remoteOrgJobHandler(org shuffle.Org, interval int) error {
// Check if it's 1 in 10 (10% chance random)
backupJob := shuffle.BackupJob{}
backupJob := shuffle.BackupJob{}
// Check if workflow backup is active
// Check if app backup is active
@@ -3647,7 +3769,6 @@ func remoteOrgJobHandler(org shuffle.Org, interval int) error {
backupJobData = []byte{}
}
syncUrl := fmt.Sprintf("%s/api/v1/cloud/sync", syncUrl)
client := shuffle.GetExternalClient(syncUrl)
req, err := http.NewRequest(
@@ -3809,7 +3930,7 @@ func runInitEs(ctx context.Context) {
}
if strings.Contains(os.Getenv("SHUFFLE_OPENSEARCH_URL"), "https") {
log.Printf("[INFO] Waiting during init to make sure the opensearch instance is up and running with security features properly")
log.Printf("[INFO] Waiting 30 seconds during init to make sure the opensearch instance is up and running with security features enabled")
time.Sleep(30 * time.Second)
}
@@ -3853,7 +3974,7 @@ func runInitEs(ctx context.Context) {
}
// FIXME: Add a randomized timer to avoid all schedules running at the same time
// Many are at 5 minutes / 1 hour. The point is to spread these out
// Many are at 5 minutes / 1 hour. The point is to spread these out
// a bit instead of all of them starting at the exact same time
//log.Printf("Schedule: %#v", schedule)
@@ -3888,22 +4009,32 @@ func runInitEs(ctx context.Context) {
log.Printf("[DEBUG] Creating org for default user %s", username)
orgId := uuid.NewV4().String()
orgSetupName := "default"
tmpOrg := shuffle.OrgMini{
Name: orgSetupName,
Id: orgId,
}
err = createNewUser(username, password, "admin", apikey, tmpOrg)
if err != nil {
log.Printf("[ERROR] Failed to create default user %s: %s", username, err)
} else {
log.Printf("[INFO] Successfully created user %s", username)
}
user, err := shuffle.GetUser(ctx, username)
newOrg := shuffle.Org{
Name: orgSetupName,
Id: orgId,
Org: orgSetupName,
Users: []shuffle.User{},
Users: []shuffle.User{*user},
Roles: []string{"admin", "user"},
CloudSync: false,
}
err = shuffle.SetOrg(ctx, newOrg, newOrg.Id)
setUsers := false
if err != nil {
log.Printf("[WARNING] Failed setting organization when creating original user: %s", err)
log.Printf("[ERROR] Failed setting organization when creating original user: %s", err)
} else {
log.Printf("[DEBUG] Successfully created the default org with id %s!", orgId)
setUsers = true
item := shuffle.Environment{
Name: defaultEnv,
@@ -3918,20 +4049,6 @@ func runInitEs(ctx context.Context) {
log.Printf("[WARNING] Failed setting up new environment")
}
}
if setUsers {
tmpOrg := shuffle.OrgMini{
Name: orgSetupName,
Id: orgId,
}
err = createNewUser(username, password, "admin", apikey, tmpOrg)
if err != nil {
log.Printf("[INFO] Failed to create default user %s: %s", username, err)
} else {
log.Printf("[INFO] Successfully created user %s", username)
}
}
}
} else {
for _, user := range users {
@@ -4145,6 +4262,8 @@ func runInitEs(ctx context.Context) {
}
}
cloneOptions = checkGitProxy(cloneOptions)
branch := os.Getenv("SHUFFLE_DOWNLOAD_AUTH_BRANCH")
if len(branch) > 0 && branch != "master" && branch != "main" {
cloneOptions.ReferenceName = plumbing.ReferenceName(branch)
@@ -4189,6 +4308,9 @@ func runInitEs(ctx context.Context) {
cloneOptions := &git.CloneOptions{
URL: apis,
}
cloneOptions = checkGitProxy(cloneOptions)
_, err = git.Clone(storer, fs, cloneOptions)
if err != nil {
log.Printf("[ERROR] Failed loading repo %s into memory: %s", apis, err)
@@ -4205,17 +4327,16 @@ func runInitEs(ctx context.Context) {
log.Printf("[INFO] Skipping download of extra API samples as %d were found", len(workflowapps))
}
if os.Getenv("SHUFFLE_HEALTHCHECK_DISABLED") != "true" {
healthcheckInterval := 30
healthcheckInterval := 30
log.Printf("[INFO] Starting healthcheck job every %d minute. Stats available on /api/v1/health/stats. Disable with SHUFFLE_HEALTHCHECK_DISABLED=true", healthcheckInterval)
job := func() {
// Prepare a fake http.responsewriter
// Prepare a fake http.responsewriter
resp := httptest.NewRecorder()
request := http.Request{}
// Add the "force=true" query to the fake request
request.URL, err = url.Parse("/api/v1/health/stats?force=true")
request.URL, err = url.Parse("/api/v1/health/stats?force=true")
if err != nil {
log.Printf("[ERROR] Failed to parse test url for healthstats: %s", err)
}
@@ -4234,7 +4355,6 @@ func runInitEs(ctx context.Context) {
log.Printf("[INFO] Finished INIT (ES)")
}
func handleVerifyCloudsync(orgId string) (shuffle.SyncFeatures, error) {
ctx := context.Background()
org, err := shuffle.GetOrg(ctx, orgId)
@@ -4813,8 +4933,6 @@ func makeWorkflowPublic(resp http.ResponseWriter, request *http.Request) {
resp.Write([]byte(fmt.Sprintf(`{"success": true}`)))
}
func handleAppZipUpload(resp http.ResponseWriter, request *http.Request) {
cors := shuffle.HandleCors(resp, request)
if cors {
@@ -4873,8 +4991,6 @@ func handleAppZipUpload(resp http.ResponseWriter, request *http.Request) {
resp.Write([]byte("OK"))
}
func initHandlers() {
var err error
ctx := context.Background()
@@ -4906,7 +5022,7 @@ func initHandlers() {
go runInitEs(ctx)
} else {
//go shuffle.runInit(ctx)
log.Printf("[ERROR] Opensearch is the only viable option. Please set SHUFFLE_ELASTIC=true")
log.Printf("[ERROR] Opensearch is the only viable option. Please set SHUFFLE_ELASTIC=true")
os.Exit(1)
}
@@ -4921,7 +5037,7 @@ func initHandlers() {
r.HandleFunc("/api/v1/users/register", handleRegister).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/users/checkusers", checkAdminLogin).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/users/getinfo", handleInfo).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/users/{userId}/apps", shuffle.HandleGetUserApps).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/users/apps", shuffle.HandleGetUserApps).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/users/generateapikey", shuffle.HandleApiGeneration).Methods("GET", "POST", "OPTIONS")
r.HandleFunc("/api/v1/users/logout", shuffle.HandleLogout).Methods("POST", "OPTIONS")
@@ -4942,6 +5058,7 @@ func initHandlers() {
r.HandleFunc("/api/v1/register", handleRegister).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/checkusers", checkAdminLogin).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/getinfo", handleInfo).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/me", handleInfo).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/getsettings", shuffle.HandleSettings).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/generateapikey", shuffle.HandleApiGeneration).Methods("GET", "POST", "OPTIONS")
r.HandleFunc("/api/v1/passwordchange", shuffle.HandlePasswordChange).Methods("POST", "OPTIONS")
@@ -4976,7 +5093,8 @@ func initHandlers() {
r.HandleFunc("/api/v1/apps/{appId}", shuffle.UpdateWorkflowAppConfig).Methods("PATCH", "OPTIONS")
r.HandleFunc("/api/v1/apps/{appId}", shuffle.DeleteWorkflowApp).Methods("DELETE", "OPTIONS")
r.HandleFunc("/api/v1/apps/{appId}/config", shuffle.GetWorkflowAppConfig).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/apps/run_hotload", handleAppHotloadRequest).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/apps/run_hotload", handleAppHotloadRequest).Methods("GET", "POST", "OPTIONS")
r.HandleFunc("/api/v1/apps/{appName}/run_hotload", handleSingleAppHotloadRequest).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/apps/get_existing", LoadSpecificApps).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/apps/download_remote", LoadSpecificApps).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/apps/validate", validateAppInput).Methods("POST", "OPTIONS")
@@ -5055,7 +5173,7 @@ func initHandlers() {
r.HandleFunc("/api/v1/triggers/gmail/register", shuffle.HandleNewGmailRegister).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/triggers/gmail/getFolders", shuffle.HandleGetGmailFolders).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/triggers/pipeline", shuffle.HandleNewPipelineRegister).Methods("POST", "OPTIONS")
//r.HandleFunc("/api/v1/triggers/pipeline/save", shuffle.HandleSavePipelineInfo).Methods("PUT", "OPTIONS")
//r.HandleFunc("/api/v1/triggers/pipeline/save", shuffle.HandleSavePipelineInfo).Methods("PUT", "OPTIONS")
r.HandleFunc("/api/v1/pipelines/{key}", handlePipelineCallback).Methods("POST", "GET", "PATCH", "PUT", "DELETE", "OPTIONS")
r.HandleFunc("/api/v1/triggers", shuffle.HandleGetTriggers).Methods("GET", "OPTIONS")
//r.HandleFunc("/api/v1/triggers/gmail/routing", handleGmailRouting).Methods("POST", "OPTIONS")
@@ -5077,12 +5195,13 @@ func initHandlers() {
//r.HandleFunc("/api/v1/orgs/", shuffle.HandleGetOrgs).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/orgs/{orgId}", shuffle.HandleGetOrg).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/orgs/{orgId}", shuffle.HandleEditOrg).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/orgs/{orgid}/forms", shuffle.HandleGetOrgForms).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/orgs/{orgId}/create_sub_org", shuffle.HandleCreateSubOrg).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/orgs/{orgId}/change", shuffle.HandleChangeUserOrg).Methods("POST", "OPTIONS") // Swaps to the org
r.HandleFunc("/api/v1/orgs/{orgId}", shuffle.HandleDeleteOrg).Methods("DELETE", "OPTIONS")
r.HandleFunc("/api/v1/orgs/{orgId}/suborgs", shuffle.HandleGetSubOrgs).Methods("GET", "OPTIONS")
// This is a new API that validates if a key has been seen before.
// Not sure what the best course of action is for it.
r.HandleFunc("/api/v1/environments/{key}/stop", shuffle.HandleStopExecutions).Methods("GET", "POST", "OPTIONS")
@@ -5090,11 +5209,13 @@ func initHandlers() {
r.HandleFunc("/api/v1/orgs/{orgId}/validate_app_values", shuffle.HandleKeyValueCheck).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/orgs/{orgId}/list_cache", shuffle.HandleListCacheKeys).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/orgs/{orgId}/cache/{cache_key}", shuffle.HandleGetCacheKey).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/orgs/{orgId}/get_cache", shuffle.HandleGetCacheKey).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/orgs/{orgId}/set_cache", shuffle.HandleSetCacheKey).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/orgs/{orgId}/delete_cache", shuffle.HandleDeleteCacheKeyPost).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/orgs/{orgId}/cache/{cache_key}", shuffle.HandleDeleteCacheKey).Methods("DELETE", "OPTIONS")
r.HandleFunc("/api/v1/orgs/{orgId}/stats", shuffle.HandleGetStatistics).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/orgs/{orgId}/stats", shuffle.HandleAppendStatistics).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/orgs/{orgId}/statistics", shuffle.HandleGetStatistics).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/orgs/{orgId}/cache", shuffle.HandleListCacheKeys).Methods("GET", "OPTIONS")
@@ -5104,7 +5225,6 @@ func initHandlers() {
r.HandleFunc("/api/v1/orgs/{orgId}/datastore", shuffle.HandleSetCacheKey).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/orgs/{orgId}/datastore/{cache_key}", shuffle.HandleDeleteCacheKey).Methods("DELETE", "OPTIONS")
// Docker orborus specific - downloads an image
r.HandleFunc("/api/v1/get_docker_image", getDockerImage).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/login_sso", shuffle.HandleSSO).Methods("GET", "POST", "OPTIONS")
@@ -5123,6 +5243,17 @@ func initHandlers() {
r.HandleFunc("/api/v1/files/{fileId}", shuffle.HandleDeleteFile).Methods("DELETE", "OPTIONS")
r.HandleFunc("/api/v1/files", shuffle.HandleGetFiles).Methods("GET", "OPTIONS")
// This structure is horrendous. Needs fixing after we got the prototype up
r.HandleFunc("/api/v1/detections/{detectionType}/connect", shuffle.HandleDetectionAutoConnect).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/detections/{detection_type}", shuffle.HandleGetDetectionRules).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/detections/{triggerId}/selected_rules", shuffle.HandleGetSelectedRules).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/detections/{triggerId}/selected_rules/save", shuffle.HandleSaveSelectedRules).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/detections/{action}", shuffle.HandleFolderToggle).Methods("PUT", "OPTIONS")
// This is weird.
r.HandleFunc("/api/v1/detections/{fileId}/{action}", shuffle.HandleToggleRule).Methods("PUT", "OPTIONS")
//r.HandleFunc("/api/v1/detections/siem/node_health", shuffle.HandleTenzirHealthUpdate).Methods("POST","OPTIONS")
// Introduced in 0.9.21 to handle notifications for e.g. failed Workflow
r.HandleFunc("/api/v1/notifications", shuffle.HandleCreateNotification).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/notifications", shuffle.HandleGetNotifications).Methods("GET", "OPTIONS")
+178 -53
View File
@@ -106,7 +106,6 @@ func createSchedule(ctx context.Context, scheduleId, workflowId, name, startNode
}
log.Printf("[INFO] Starting frequency for execution: %d", newfrequency)
//jobret, err := newscheduler.Every(newfrequency).Seconds().NotImmediately().Run(job)
jobret, err := newscheduler.Every(newfrequency).Seconds().Run(job)
@@ -292,29 +291,44 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) {
ctx := shuffle.GetContext(request)
env, err := shuffle.GetEnvironment(ctx, orgId, "")
timeNow := time.Now().Unix()
if err == nil && len(env.Id) > 0 && len(env.Name) > 0 {
if err == nil && len(env.Id) > 0 && len(env.Name) > 0 && request.Method == "POST" {
// Updates every 60 seconds~
if time.Now().Unix() > env.Edited+60 {
env.RunningIp = shuffle.GetRequestIp(request)
// Orborus label = custom label for Orborus
if len(orborusLabel) > 0 {
env.RunningIp = orborusLabel
}
if request.Method == "POST" {
body, err := ioutil.ReadAll(request.Body)
if err == nil {
var envData shuffle.OrborusStats
err = json.Unmarshal(body, &envData)
if err == nil {
if envData.Swarm {
env.Licensed = true
env.RunType = "docker"
}
// Set the checkin cache
if envData.Kubernetes {
env.RunType = "k8s"
}
body, err := ioutil.ReadAll(request.Body)
if err == nil {
var envData shuffle.OrborusStats
err = json.Unmarshal(body, &envData)
if err == nil {
envData.RunningIp = env.RunningIp
marshalled, err := json.Marshal(envData)
if err == nil {
cacheKey := fmt.Sprintf("queueconfig-%s-%s", env.Name, env.OrgId)
go shuffle.SetCache(context.Background(), cacheKey, marshalled, 2)
}
if envData.Swarm {
env.Licensed = true
env.RunType = "docker"
}
if envData.Kubernetes {
env.RunType = "k8s"
}
envData.DataLake = env.DataLake
}
}
@@ -572,14 +586,20 @@ func handleGetStreamResults(resp http.ResponseWriter, request *http.Request) {
//return
}
if len(actionResult.ExecutionId) == 0 {
resp.WriteHeader(400)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Provide execution_id and authorization"}`)))
return
}
ctx := context.Background()
workflowExecution, err := shuffle.GetWorkflowExecution(ctx, actionResult.ExecutionId)
if err != nil {
if err != nil || workflowExecution.ExecutionId != actionResult.ExecutionId {
if len(actionResult.ExecutionId) > 0 {
log.Printf("[WARNING][%s] Failed getting execution (streamresult): %s", actionResult.ExecutionId, err)
}
resp.WriteHeader(401)
resp.WriteHeader(400)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Bad authorization key or execution_id might not exist."}`)))
return
}
@@ -638,9 +658,27 @@ func handleGetStreamResults(resp http.ResponseWriter, request *http.Request) {
}
}
if workflowExecution.Workflow.Sharing == "form" {
newWorkflow := shuffle.Workflow{
Name: workflowExecution.Workflow.Name,
ID: workflowExecution.Workflow.ID,
Owner: workflowExecution.Workflow.Owner,
OrgId: workflowExecution.Workflow.OrgId,
Sharing: workflowExecution.Workflow.Sharing,
Description: workflowExecution.Workflow.Description,
InputQuestions: workflowExecution.Workflow.InputQuestions,
FormControl: workflowExecution.Workflow.FormControl,
}
workflowExecution.Results = []shuffle.ActionResult{}
workflowExecution.Workflow = newWorkflow
}
newjson, err := json.Marshal(workflowExecution)
if err != nil {
resp.WriteHeader(401)
resp.WriteHeader(500)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed unpacking workflow execution"}`)))
return
}
@@ -670,7 +708,7 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
}
//log.Printf("Actionresult unmarshal: %s", string(body))
log.Printf("[DEBUG] Got workflow result from %s of length %d", request.RemoteAddr, len(body))
//log.Printf("[DEBUG] Got workflow result from %s of length %d", request.RemoteAddr, len(body))
ctx := context.Background()
err = shuffle.ValidateNewWorkerExecution(ctx, body)
if err == nil {
@@ -681,7 +719,7 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
log.Printf("[DEBUG] Handling other execution variant (subflow?): %s", err)
}
log.Printf("[DEBUG] Got workflow result from %s of length %d.", request.RemoteAddr, len(body))
//log.Printf("[DEBUG] Got workflow result from %s of length %d.", request.RemoteAddr, len(body))
var actionResult shuffle.ActionResult
err = json.Unmarshal(body, &actionResult)
@@ -739,8 +777,7 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
// Will make sure transactions are always ran for an execution. This is recursive if it fails. Allowed to fail up to 5 times
func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workflowExecutionId string, actionResult shuffle.ActionResult, resp http.ResponseWriter) {
log.Printf("[DEBUG][%s] Running workflow execution update", workflowExecutionId)
log.Printf("[DEBUG][%s] Running workflow execution update with result from %s (%s) of status %s", workflowExecutionId, actionResult.Action.Label, actionResult.Action.ID, actionResult.Status)
// Should start a tx for the execution here
workflowExecution, err := shuffle.GetWorkflowExecution(ctx, workflowExecutionId)
@@ -771,7 +808,6 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl
setExecution := true
if setExecution || workflowExecution.Status == "FINISHED" || workflowExecution.Status == "ABORTED" || workflowExecution.Status == "FAILURE" {
err = shuffle.SetWorkflowExecution(ctx, *workflowExecution, true)
//err = shuffle.SetWorkflowExecution(ctx, *workflowExecution, dbSave)
if err != nil {
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed setting workflowexecution actionresult: %s"}`, err)))
@@ -925,7 +961,7 @@ func deleteWorkflow(resp http.ResponseWriter, request *http.Request) {
if len(workflow.ParentWorkflowId) > 0 {
resp.WriteHeader(403)
resp.Write([]byte(`{"success": false, "reason": "Can't delete a workflow distributed from your parent org"}`))
return
return
}
if user.Id != workflow.Owner || len(user.Id) == 0 {
@@ -939,6 +975,27 @@ func deleteWorkflow(resp http.ResponseWriter, request *http.Request) {
}
}
// Look for Child workflows and delete them
if workflow.ParentWorkflowId == "" {
log.Printf("[DEBUG] Looking for child workflows for workflow %s to delete. User %s (%s) in org %s (%s)", workflow.ID, user.Username, user.Id, user.ActiveOrg.Name, user.ActiveOrg.Id)
childWorkflows, err := shuffle.ListChildWorkflows(ctx, workflow.ID)
if err != nil {
log.Printf("[ERROR] Failed to list child workflows: %s", err)
} else {
log.Printf("\n\n[DEBUG] Found %d child workflows for workflow %s\n\n", len(childWorkflows), workflow.ID)
// Find cookies and append them to request.Header to replicate current request as closely as possible
for _, childWorkflow := range childWorkflows {
if childWorkflow.ID == workflow.ID {
continue
}
go shuffle.SendDeleteWorkflowRequest(childWorkflow, request)
}
}
}
// Clean up triggers and executions
for _, item := range workflow.Triggers {
if item.TriggerType == "SCHEDULE" && item.Status != "uninitialized" {
@@ -984,8 +1041,6 @@ func deleteWorkflow(resp http.ResponseWriter, request *http.Request) {
resp.Write([]byte(`{"success": true}`))
}
func handleExecution(id string, workflow shuffle.Workflow, request *http.Request, orgId string) (shuffle.WorkflowExecution, string, error) {
//go func() {
// log.Printf("\n\nPRE TIME: %s\n\n", time.Now().Format("2006-01-02 15:04:05"))
@@ -1004,17 +1059,6 @@ func handleExecution(id string, workflow shuffle.Workflow, request *http.Request
workflow = *tmpworkflow
}
/*
if len(workflow.ExecutingOrg.Id) == 0 {
if len(orgId) > 0 {
workflow.ExecutingOrg.Id = orgId
} else {
log.Printf("[INFO] Stopped execution because there is no executing org for workflow %s", workflow.ID)
return shuffle.WorkflowExecution{}, fmt.Sprintf("Workflow has no executing org defined"), errors.New("Workflow has no executing org defined")
}
}
*/
if len(workflow.Actions) == 0 {
workflow.Actions = []shuffle.Action{}
} else {
@@ -1065,28 +1109,31 @@ func handleExecution(id string, workflow shuffle.Workflow, request *http.Request
workflowExecution, execInfo, _, workflowExecErr := shuffle.PrepareWorkflowExecution(ctx, workflow, request, int64(maxExecutionDepth))
if workflowExecErr != nil {
err := shuffle.SetWorkflowExecution(ctx, workflowExecution, true)
if err != nil {
log.Printf("[ERROR] Failed setting workflow execution during init (2): %s", err)
if len(workflowExecution.Workflow.Actions) > 0 && len(workflowExecution.Results) > 0 && len(workflowExecution.ExecutionId) > 0 {
err := shuffle.SetWorkflowExecution(ctx, workflowExecution, true)
if err != nil {
log.Printf("[ERROR] Failed setting workflow execution during init (2): %s", err)
}
}
if strings.Contains(fmt.Sprintf("%s", workflowExecErr), "User Input") {
// Special for user input callbacks
log.Printf("[INFO] User input callback: %s", workflowExecErr)
// return workflowExecution, fmt.Sprintf("%s", err), nil
//log.Printf("[INFO] User input callback: %s", workflowExecErr)
return shuffle.WorkflowExecution{}, "", nil
} else {
log.Printf("[ERROR] Failed in prepareExecution: '%s'", err)
return shuffle.WorkflowExecution{}, fmt.Sprintf("Failed running: %s", err), err
log.Printf("[ERROR] Failed in prepareExecution: '%s'", workflowExecErr)
return shuffle.WorkflowExecution{}, fmt.Sprintf("Failed running: %s", workflowExecErr), workflowExecErr
}
}
err := imageCheckBuilder(execInfo.ImageNames)
if err != nil {
log.Printf("[ERROR] Failed building the required images from %#v: %s", execInfo.ImageNames, err)
return shuffle.WorkflowExecution{}, "Failed unmarshal during execution", err
}
/*
makeNew := true
start, startok := request.URL.Query()["start"]
if request.Method == "POST" {
@@ -1206,7 +1253,7 @@ func handleExecution(id string, workflow shuffle.Workflow, request *http.Request
answer, answerok := request.URL.Query()["answer"]
referenceId, referenceok := request.URL.Query()["reference_execution"]
if answerok && referenceok {
if answerok && referenceok && len(answer) > 0 && len(referenceId) > 0 {
// If answer is false, reference execution with result
log.Printf("[INFO] Answer is OK AND reference is OK!")
if answer[0] == "false" {
@@ -1230,7 +1277,7 @@ func handleExecution(id string, workflow shuffle.Workflow, request *http.Request
log.Printf("%s - %s", result.Action.ID, start[0])
if result.Action.ID == start[0] {
note, noteok := request.URL.Query()["note"]
if noteok {
if noteok && len(note) > 0 {
result.Result = fmt.Sprintf("User note: %s", note[0])
} else {
result.Result = fmt.Sprintf("User clicked %s", answer[0])
@@ -1354,7 +1401,7 @@ func handleExecution(id string, workflow shuffle.Workflow, request *http.Request
}
}
childNodes := shuffle.FindChildNodes(workflowExecution, workflowExecution.Start, []string{}, []string{})
childNodes := shuffle.FindChildNodes(workflowExecution.Workflow, workflowExecution.Start, []string{}, []string{})
startFound := false
newActions := []shuffle.Action{}
@@ -1560,7 +1607,6 @@ func handleExecution(id string, workflow shuffle.Workflow, request *http.Request
// newTriggers = append(newTriggers, trigger)
//}
//workflowExecution.Workflow.Triggers = newTriggers
_ = removeTriggers
if !startFound {
if len(workflowExecution.Start) == 0 && len(workflowExecution.Workflow.Start) > 0 {
@@ -1585,12 +1631,17 @@ func handleExecution(id string, workflow shuffle.Workflow, request *http.Request
if len(workflowExecution.ExecutionOrg) == 0 && len(workflow.ExecutingOrg.Id) > 0 {
workflowExecution.ExecutionOrg = workflow.ExecutingOrg.Id
}
*/
//workflowExecution, execInfo, _, workflowExecErr := shuffle.PrepareWorkflowExecution(ctx, workflow, request, int64(maxExecutionDepth))
err = shuffle.SetWorkflowExecution(ctx, workflowExecution, true)
if err != nil {
log.Printf("[ERROR] Failed setting workflow execution during init (2): %s", err)
}
onpremExecution := execInfo.OnpremExecution
_ = onpremExecution
environments := execInfo.Environments
var allEnvs []shuffle.Environment
if len(workflowExecution.ExecutionOrg) > 0 {
//log.Printf("[INFO] Executing ORG: %s", workflowExecution.ExecutionOrg)
@@ -1854,7 +1905,6 @@ func executeWorkflow(resp http.ResponseWriter, request *http.Request) {
}
log.Printf("[INFO] Inside execute workflow for ID %s", fileId)
ctx := context.Background()
workflow, err := shuffle.GetWorkflow(ctx, fileId)
if err != nil && workflow.ID == "" {
@@ -2408,7 +2458,7 @@ func scheduleWorkflow(resp http.ResponseWriter, request *http.Request) {
return
}
workflow.Schedules = append(workflow.Schedules, schedule)
//workflow.Schedules = append(workflow.Schedules, schedule)
err = shuffle.SetWorkflow(ctx, *workflow, workflow.ID)
if err != nil {
log.Printf("Failed setting workflow for schedule: %s", err)
@@ -2665,6 +2715,8 @@ func loadGithubWorkflows(url, username, password, userId, branch, orgId string)
cloneOptions.ReferenceName = plumbing.ReferenceName(branch)
}
cloneOptions = checkGitProxy(cloneOptions)
storer := memory.NewStorage()
r, err := git.Clone(storer, fs, cloneOptions)
if err != nil {
@@ -2766,6 +2818,70 @@ func loadSpecificWorkflows(resp http.ResponseWriter, request *http.Request) {
resp.Write([]byte(fmt.Sprintf(`{"success": true}`)))
}
func handleSingleAppHotloadRequest(resp http.ResponseWriter, request *http.Request) {
cors := shuffle.HandleCors(resp, request)
if cors {
return
}
ctx := context.Background()
cacheKey := fmt.Sprintf("workflowapps-sorted-1000")
shuffle.DeleteCache(ctx, cacheKey)
cacheKey = fmt.Sprintf("workflowapps-sorted-500")
shuffle.DeleteCache(ctx, cacheKey)
cacheKey = fmt.Sprintf("workflowapps-sorted-0")
shuffle.DeleteCache(ctx, cacheKey)
// Just need to be logged in
// FIXME - should have some permissions?
user, err := shuffle.HandleApiAuthentication(resp, request)
if err != nil {
log.Printf("Api authentication failed in app hotload: %s", err)
resp.WriteHeader(401)
resp.Write([]byte(`{"success": false}`))
return
}
if user.Role != "admin" {
resp.WriteHeader(401)
resp.Write([]byte(`{"success": false, "reason": "Must be admin to hotload apps"}`))
return
}
location := os.Getenv("SHUFFLE_APP_HOTLOAD_FOLDER")
if len(location) == 0 {
resp.WriteHeader(500)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "SHUFFLE_APP_HOTLOAD_FOLDER not specified in .env"}`)))
return
}
requestUrlFields := strings.Split(request.URL.String(), "/")
var appName string
if requestUrlFields[1] == "api" {
if len(requestUrlFields) <= 4 {
resp.WriteHeader(401)
resp.Write([]byte(`{"success": false}`))
return
}
appName = requestUrlFields[4]
if strings.Contains(appName, "?") {
appName = strings.Split(appName, "?")[0]
}
}
location = location + "/" + appName
log.Printf("[INFO] Starting hotloading from %s", location)
err = handleAppHotload(ctx, location, true)
if err != nil {
log.Printf("[WARNING] Failed app hotload: %s", err)
resp.WriteHeader(500)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err)))
return
}
cacheKey = fmt.Sprintf("workflowapps-sorted-100")
shuffle.DeleteCache(ctx, cacheKey)
cacheKey = fmt.Sprintf("workflowapps-sorted-500")
shuffle.DeleteCache(ctx, cacheKey)
cacheKey = fmt.Sprintf("workflowapps-sorted-1000")
shuffle.DeleteCache(ctx, cacheKey)
resp.WriteHeader(200)
resp.Write([]byte(fmt.Sprintf(`{"success": true}`)))
}
func handleAppHotloadRequest(resp http.ResponseWriter, request *http.Request) {
cors := shuffle.HandleCors(resp, request)
if cors {
@@ -3374,7 +3490,15 @@ func executeSingleAction(resp http.ResponseWriter, request *http.Request) {
}
ctx := context.Background()
workflowExecution, err := shuffle.PrepareSingleAction(ctx, user, fileId, body)
runValidationAction := false
query := request.URL.Query()
validation, ok := query["validation"]
if ok && validation[0] == "true" {
runValidationAction = true
}
workflowExecution, err := shuffle.PrepareSingleAction(ctx, user, fileId, body, runValidationAction)
if err != nil {
log.Printf("[INFO] Failed workflowrequest POST read: %s", err)
resp.WriteHeader(401)
@@ -3410,7 +3534,7 @@ func executeSingleAction(resp http.ResponseWriter, request *http.Request) {
// FIXME: Should use environment that is in the source workflow if it exists
for i, _ := range workflowExecution.Workflow.Actions {
workflowExecution.Workflow.Actions[i].Environment = environment
workflowExecution.Workflow.Actions[i].Label = "TMP"
workflowExecution.Workflow.Actions[i].Label = "TMP"
}
shuffle.SetWorkflowExecution(ctx, workflowExecution, false)
@@ -3940,6 +4064,8 @@ func LoadSpecificApps(resp http.ResponseWriter, request *http.Request) {
}
}
cloneOptions = checkGitProxy(cloneOptions)
storer := memory.NewStorage()
r, err := git.Clone(storer, fs, cloneOptions)
if err != nil {
@@ -4193,7 +4319,6 @@ func checkUnfinishedExecution(resp http.ResponseWriter, request *http.Request) {
log.Printf("[ERROR] Failed adding execution to db: %s", err)
}
resp.WriteHeader(200)
resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "Reran workflow in %s"}`, parsedEnv)))