diff --git a/db-connector.go b/db-connector.go index d4d42de2..4307343a 100755 --- a/db-connector.go +++ b/db-connector.go @@ -6140,12 +6140,18 @@ func GetOpenApiDatastore(ctx context.Context, id string) (ParsedOpenApi, error) } // Index = Username -func SetSession(ctx context.Context, user User, value string) error { +func SetSession(ctx context.Context, user *User, value string) error { //parsedKey := strings.ToLower(user.Username) // Non indexed User data parsedKey := user.Id user.Session = value + now := time.Now().Unix() + if user.SessionCreatedAt == 0 { + user.SessionCreatedAt = now + } + user.SessionLastActivityAt = now + nameKey := "Users" if project.DbType == "opensearch" { data, err := json.Marshal(user) diff --git a/shared.go b/shared.go index 1bf11c93..28720622 100644 --- a/shared.go +++ b/shared.go @@ -3,6 +3,7 @@ package shuffle import ( "bytes" "context" + "crypto/sha256" "crypto/tls" "crypto/x509" "errors" @@ -16,18 +17,17 @@ import ( "os" "path/filepath" "reflect" - "crypto/sha256" - "sync" "hash/fnv" neturl "net/url" "path" "sort" + "sync" "unicode" - openai "github.com/sashabaranov/go-openai" "github.com/go-git/go-billy/v5" "github.com/go-git/go-billy/v5/memfs" + openai "github.com/sashabaranov/go-openai" "google.golang.org/api/cloudfunctions/v1" "google.golang.org/api/googleapi" "google.golang.org/api/iterator" @@ -79,8 +79,9 @@ import ( "golang.org/x/crypto/bcrypt" "golang.org/x/oauth2" - "github.com/Masterminds/semver" "runtime" + + "github.com/Masterminds/semver" dockerclient "github.com/docker/docker/client" ) @@ -287,6 +288,34 @@ func isLoop(arg string) bool { return false } +func getSessionIdleTimeout() time.Duration { + val := os.Getenv("SHUFFLE_SESSION_IDLE_TIMEOUT") + if val == "" { + return 3600 * time.Second + } + seconds, err := strconv.Atoi(val) + if err != nil || seconds <= 0 { + return 3600 * time.Second + } + return time.Duration(seconds) * time.Second +} + +func getSessionMaxLifetime() time.Duration { + val := os.Getenv("SHUFFLE_SESSION_MAX_LIFETIME") + if val == "" { + return 0 + } + seconds, err := strconv.Atoi(val) + if err != nil || seconds <= 0 { + return 0 + } + return time.Duration(seconds) * time.Second +} + +func getSessionExpiration() time.Time { + return time.Now().Add(getSessionIdleTimeout()) +} + func ConstructSessionCookie(value string, expires time.Time) *http.Cookie { c := http.Cookie{ Name: "session_token", @@ -329,6 +358,23 @@ func constructSessionDeleteCookie() *http.Cookie { return c } +// expireSession clears the session on both the server (DB, cache) and client (cookie). +func expireSession(ctx context.Context, resp http.ResponseWriter, user *User, reason string) error { + if resp != nil { + newCookie := constructSessionDeleteCookie() + http.SetCookie(resp, newCookie) + newCookie.Name = "__session" + http.SetCookie(resp, newCookie) + } + go DeleteCache(ctx, fmt.Sprintf("session_%s", user.Session)) + user.Session = "" + user.SessionCreatedAt = 0 + user.SessionLastActivityAt = 0 + user.ValidatedSessionOrgs = []string{} + go SetUser(ctx, user, false) + return errors.New(reason) +} + func HandleSet2fa(resp http.ResponseWriter, request *http.Request) { cors := HandleCors(resp, request) if cors { @@ -568,7 +614,7 @@ func HandleSet2fa(resp http.ResponseWriter, request *http.Request) { if len(user.Session) != 0 { log.Printf("[INFO] User session exists - resetting session") - expiration := time.Now().Add(8 * time.Hour) + expiration := getSessionExpiration() newCookie := ConstructSessionCookie(user.Session, expiration) @@ -578,25 +624,12 @@ func HandleSet2fa(resp http.ResponseWriter, request *http.Request) { http.SetCookie(resp, newCookie) //log.Printf("SESSION LENGTH MORE THAN 0 IN LOGIN: %s", user.Session) - returnValue.Cookies = append(returnValue.Cookies, SessionCookie{ - Key: "session_token", - Value: user.Session, - Expiration: expiration.Unix(), - }) - - returnValue.Cookies = append(returnValue.Cookies, SessionCookie{ - Key: "__session", - Value: user.Session, - Expiration: expiration.Unix(), - }) - - loginData = fmt.Sprintf(`{"success": true, "cookies": [{"key": "session_token", "value": "%s", "expiration": %d}]}`, user.Session, expiration.Unix()) newData, err := json.Marshal(returnValue) if err == nil { loginData = string(newData) } - err = SetSession(ctx, user, user.Session) + err = SetSession(ctx, &user, user.Session) if err != nil { log.Printf("[WARNING] Error adding session to database: %s", err) } else { @@ -620,7 +653,7 @@ func HandleSet2fa(resp http.ResponseWriter, request *http.Request) { log.Printf("[INFO] User session for %s (%s) is empty - create one!", user.Username, user.Id) sessionToken := uuid.NewV4().String() - expiration := time.Now().Add(8 * time.Hour) + expiration := getSessionExpiration() newCookie := ConstructSessionCookie(sessionToken, expiration) // Does it not set both? @@ -630,24 +663,13 @@ func HandleSet2fa(resp http.ResponseWriter, request *http.Request) { http.SetCookie(resp, newCookie) // ADD TO DATABASE - err = SetSession(ctx, user, sessionToken) + err = SetSession(ctx, &user, sessionToken) if err != nil { log.Printf("[DEBUG] Error adding session to database: %s", err) } user.Session = sessionToken - returnValue.Cookies = append(returnValue.Cookies, SessionCookie{ - Key: "session_token", - Value: sessionToken, - Expiration: expiration.Unix(), - }) - - returnValue.Cookies = append(returnValue.Cookies, SessionCookie{ - Key: "__session", - Value: sessionToken, - Expiration: expiration.Unix(), - }) user.MFA = foundUser.MFA err = SetUser(ctx, &user, true) if err != nil { @@ -657,7 +679,6 @@ func HandleSet2fa(resp http.ResponseWriter, request *http.Request) { return } - loginData = fmt.Sprintf(`{"success": true, "cookies": [{"key": "session_token", "value": "%s", "expiration": %d}]}`, sessionToken, expiration.Unix()) newData, err := json.Marshal(returnValue) if err == nil { loginData = string(newData) @@ -1781,6 +1802,8 @@ func HandleLogout(resp http.ResponseWriter, request *http.Request) { userInfo.UsersLastSession = userInfo.Session userInfo.Session = "" + userInfo.SessionCreatedAt = 0 + userInfo.SessionLastActivityAt = 0 userInfo.ValidatedSessionOrgs = []string{} err := SetUser(ctx, &userInfo, false) if err != nil { @@ -1915,7 +1938,7 @@ func GetAppAuthentication(resp http.ResponseWriter, request *http.Request) { newAuthField := auth for index, _ := range auth.Fields { - // Allowing these fields specifically, as they typically aren't + // Allowing these fields specifically, as they typically aren't // sensitive, and the API is authenticated. if auth.Fields[index].Key == "url" || auth.Fields[index].Key == "model" { @@ -2382,23 +2405,23 @@ func AddAppAuthentication(resp http.ResponseWriter, request *http.Request) { // Removing as being strict on EXTRA fields don't matter much // Apps can handle this anyway /* - // Check if the items are correct - for _, field := range appAuth.Fields { - found := false - for _, param := range app.Authentication.Parameters { - //log.Printf("Fields: %s - %s", field, param.Name) - if field.Key == param.Name { - found = true + // Check if the items are correct + for _, field := range appAuth.Fields { + found := false + for _, param := range app.Authentication.Parameters { + //log.Printf("Fields: %s - %s", field, param.Name) + if field.Key == param.Name { + found = true + } } - } - if !found { - log.Printf("[WARNING] Failed finding field '%s' in appauth fields for %s", field.Key, appAuth.App.Name) - resp.WriteHeader(409) - resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "All auth fields required"}`))) - return + if !found { + log.Printf("[WARNING] Failed finding field '%s' in appauth fields for %s", field.Key, appAuth.App.Name) + resp.WriteHeader(409) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "All auth fields required"}`))) + return + } } - } */ } } @@ -3336,7 +3359,7 @@ func HandleGetEnvironments(resp http.ResponseWriter, request *http.Request) { if len(environments) == 0 { resp.WriteHeader(404) resp.Write([]byte(`{"success": false, "reason": "Can't find environment. Does it exist?"}`)) - return + return } } @@ -3727,7 +3750,7 @@ func HandleApiAuthentication(resp http.ResponseWriter, request *http.Request) (U if err != nil { // Due to execution auth if !strings.Contains(request.URL.String(), "authorization=") && !strings.Contains(request.URL.String(), "execution_id=") { - if debug { + if debug { log.Printf("[DEBUG] Apikey '%s' doesn't exist. URL: %#v: %s", apikeyCheck[1], request.URL.String(), err) } } @@ -3829,7 +3852,7 @@ func HandleApiAuthentication(resp http.ResponseWriter, request *http.Request) (U } else { // Check if both session tokens are set // Compatibility issues - //expiration := time.Now().Add(8 * time.Hour) + //expiration := getSessionExpiration() newCookie := ConstructSessionCookie(sessionToken, c.Expires) newCookie.MaxAge = c.MaxAge @@ -3904,6 +3927,39 @@ func HandleApiAuthentication(resp http.ResponseWriter, request *http.Request) (U user.SessionLogin = true + // ── Session timeout enforcement ────────────────────────────────── + now := time.Now() + idleTimeout := getSessionIdleTimeout() + maxLifetime := getSessionMaxLifetime() + + if user.Session != "" { + // Idle timeout: expire if the user hasn't made any request within the configured window + if user.SessionLastActivityAt > 0 && now.Unix()-user.SessionLastActivityAt > int64(idleTimeout.Seconds()) { + return User{}, expireSession(ctx, resp, &user, "Session expired due to inactivity") + } + + // Max lifetime: expire if the session has existed longer than the absolute maximum, + // regardless of activity. Only enforced when SHUFFLE_SESSION_MAX_LIFETIME is set. + if maxLifetime > 0 && user.SessionCreatedAt > 0 && now.Unix()-user.SessionCreatedAt > int64(maxLifetime.Seconds()) { + return User{}, expireSession(ctx, resp, &user, "Session max lifetime exceeded") + } + + // Refresh the session after every successful authenciation, but at most once every 60 seconds per session, + // to avoid hammering Opensearch on every keystroke or rapid-fire API call. + if now.Unix()-user.SessionLastActivityAt >= 60 { + user.SessionLastActivityAt = now.Unix() + go SetUser(ctx, &user, false) + + // Slide the cookie's Expires forward so the browser also enforces the idle timeout at the HTTP level. + if resp != nil { + refreshedCookie := ConstructSessionCookie(user.Session, getSessionExpiration()) + http.SetCookie(resp, refreshedCookie) + refreshedCookie.Name = "__session" + http.SetCookie(resp, refreshedCookie) + } + } + } + // Means session exists, but return user, nil } @@ -4563,7 +4619,7 @@ func GetWorkflowExecutionsV2(resp http.ResponseWriter, request *http.Request) { cursor := "" cursorList, cursorOk := request.URL.Query()["cursor"] - if cursorOk && len(cursorList) > 60{ + if cursorOk && len(cursorList) > 60 { cursor = cursorList[0] } @@ -12940,7 +12996,7 @@ func HandleChangeUserOrg(resp http.ResponseWriter, request *http.Request) { return } - expiration := time.Now().Add(8 * time.Hour) + expiration := getSessionExpiration() newCookie := ConstructSessionCookie(user.Session, expiration) http.SetCookie(resp, newCookie) @@ -16650,26 +16706,10 @@ func HandleLogin(resp http.ResponseWriter, request *http.Request) { } } - regionUrl := "" - if project.Environment == "cloud" { - if len(userdata.ActiveOrg.RegionUrl) > 0 { - regionUrl = userdata.ActiveOrg.RegionUrl - } else { - org, err := GetOrg(ctx, userdata.ActiveOrg.Id) - if err != nil { - log.Printf("[ERROR] Failed getting org %s during login for %s (%s): %s", userdata.ActiveOrg.Id, userdata.Username, userdata.Id, err) - } else { - if strings.Contains(strings.ToLower(org.RegionUrl), "http") { - regionUrl = strings.ToLower(org.RegionUrl) - } - } - } - } - // Had to set this due to session hashing rollback if len(userdata.Session) != 0 && len(userdata.Session) == 36 && !changeActiveOrg { log.Printf("[INFO] User session exists - resetting session") - expiration := time.Now().Add(8 * time.Hour) + expiration := getSessionExpiration() newCookie := ConstructSessionCookie(userdata.Session, expiration) http.SetCookie(resp, newCookie) @@ -16677,19 +16717,6 @@ func HandleLogin(resp http.ResponseWriter, request *http.Request) { newCookie.Name = "__session" http.SetCookie(resp, newCookie) - //log.Printf("SESSION LENGTH MORE THAN 0 IN LOGIN: %s", userdata.Session) - returnValue.Cookies = append(returnValue.Cookies, SessionCookie{ - Key: "session_token", - Value: userdata.Session, - Expiration: expiration.Unix(), - }) - - returnValue.Cookies = append(returnValue.Cookies, SessionCookie{ - Key: "__session", - Value: userdata.Session, - Expiration: expiration.Unix(), - }) - // Singul handler if project.Environment == "cloud" { newCookie.Name = "__session" @@ -16705,13 +16732,12 @@ func HandleLogin(resp http.ResponseWriter, request *http.Request) { http.SetCookie(resp, newCookie) } - loginData = fmt.Sprintf(`{"success": true, "cookies": [{"key": "session_token", "value": "%s", "expiration": %d}], "region_url": "%s"}`, userdata.Session, expiration.Unix(), regionUrl) newData, err := json.Marshal(returnValue) if err == nil { loginData = string(newData) } - err = SetSession(ctx, userdata, userdata.Session) + err = SetSession(ctx, &userdata, userdata.Session) if err != nil { log.Printf("[WARNING] Error adding session to database: %s", err) } else { @@ -16733,7 +16759,7 @@ func HandleLogin(resp http.ResponseWriter, request *http.Request) { log.Printf("[INFO] User session for %s (%s) is empty - create one!", userdata.Username, userdata.Id) sessionToken := uuid.NewV4().String() - expiration := time.Now().Add(8 * time.Hour) + expiration := getSessionExpiration() newCookie := ConstructSessionCookie(sessionToken, expiration) // Does it not set both? @@ -16743,25 +16769,13 @@ func HandleLogin(resp http.ResponseWriter, request *http.Request) { http.SetCookie(resp, newCookie) // ADD TO DATABASE - err = SetSession(ctx, userdata, sessionToken) + err = SetSession(ctx, &userdata, sessionToken) if err != nil { log.Printf("[DEBUG] Error adding session to database: %s", err) } userdata.Session = sessionToken - returnValue.Cookies = append(returnValue.Cookies, SessionCookie{ - Key: "session_token", - Value: sessionToken, - Expiration: expiration.Unix(), - }) - - returnValue.Cookies = append(returnValue.Cookies, SessionCookie{ - Key: "__session", - Value: sessionToken, - Expiration: expiration.Unix(), - }) - // Singul handler if project.Environment == "cloud" { newCookie.Name = "__session" @@ -16781,7 +16795,6 @@ func HandleLogin(resp http.ResponseWriter, request *http.Request) { return } - loginData = fmt.Sprintf(`{"success": true, "cookies": [{"key": "session_token", "value": "%s", "expiration": %d}], "region_url": "%s"}`, sessionToken, expiration.Unix(), regionUrl) newData, err := json.Marshal(returnValue) if err == nil { loginData = string(newData) @@ -18222,61 +18235,61 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut // Special handler for AI Agent -> App run setCache := true - if skipAgentWait == "true" && actionResult.Action.AppName == "openai" && len(workflowExecution.ExecutionParent) > 0 { + if skipAgentWait == "true" && actionResult.Action.AppName == "openai" && len(workflowExecution.ExecutionParent) > 0 { foundParentExec, err := GetWorkflowExecution(ctx, workflowExecution.ExecutionParent) - if err != nil || len(foundParentExec.ExecutionId) == 0 { + if err != nil || len(foundParentExec.ExecutionId) == 0 { log.Printf("[ERROR][%s] Failed to find AI Parent exec %s", workflowExecution.ExecutionId, workflowExecution.ExecutionParent) } else { // Question: How does it know where to send it? - // None of these methods work. + // None of these methods work. // Since it's based on Parent => Node, it should be ExecutionSourceNode being the AI one // ExecutionSourceNode string `json:"execution_source_node" yaml:"execution_source_node"` startNode := Action{} - if len(workflowExecution.ExecutionSourceNode) == 0 { + if len(workflowExecution.ExecutionSourceNode) == 0 { log.Printf("[ERROR][%s] Agent run is missing ExecutionSourceNode from parent execution %s", workflowExecution.ExecutionId, foundParentExec.ExecutionId) } else { // This doesn't work due to e.g. having multiple nodes in the same one // AKA it's guessing - for _, action := range foundParentExec.Workflow.Actions { - if action.ID == workflowExecution.ExecutionSourceNode { + for _, action := range foundParentExec.Workflow.Actions { + if action.ID == workflowExecution.ExecutionSourceNode { startNode = action break } } } - if startNode.Name != "" { + if startNode.Name != "" { skipAgentContinue := false - if strings.Contains(actionResult.Result, "success") { + if strings.Contains(actionResult.Result, "success") { quickUnmarshal := ResultChecker{} err := json.Unmarshal([]byte(actionResult.Result), &quickUnmarshal) if err == nil && quickUnmarshal.Success == false { skipAgentContinue = true oldAgentOutput := AgentOutput{} foundError := fmt.Sprintf("LLM received call failed from app: ") - if len(quickUnmarshal.Reason) > 0 { + if len(quickUnmarshal.Reason) > 0 { foundError += fmt.Sprintf(quickUnmarshal.Reason) } - // Tries to map it in from the openai request + // Tries to map it in from the openai request if len(oldAgentOutput.OriginalInput) == 0 { - for _, param := range actionResult.Action.Parameters { - if param.Name != "body" { + for _, param := range actionResult.Action.Parameters { + if param.Name != "body" { continue } - // Marshal into openai conversation request + // Marshal into openai conversation request openaiReq := openai.ChatCompletionRequest{} unmarshalledErr := json.Unmarshal([]byte(param.Value), &openaiReq) if unmarshalledErr != nil { log.Printf("[ERROR] Failed unmarshalling body into openai request: %s", unmarshalledErr) break - } + } if len(openaiReq.Messages) > 0 { for _, userMessage := range openaiReq.Messages { - if !strings.HasPrefix(userMessage.Content, "USER REQUEST:") { + if !strings.HasPrefix(userMessage.Content, "USER REQUEST:") { continue } @@ -18293,13 +18306,13 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut } } - if !skipAgentContinue { + if !skipAgentContinue { callerName := "LLMResponse" marshalledResult, err := json.Marshal(actionResult) - if err != nil { + if err != nil { log.Printf("[ERROR] AI Agent (10): Failed marshalling actionResult: %s", err) } else { - go HandleAiAgentExecutionStart(*foundParentExec, startNode, false, callerName, marshalledResult) + go HandleAiAgentExecutionStart(*foundParentExec, startNode, false, callerName, marshalledResult) } } } else { @@ -18525,12 +18538,12 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut workflowExecution.ExecutionVariables = append(workflowExecution.ExecutionVariables, actionResult.Action.ExecutionVariable) } // @yashsinghcodes: Something to force the executionVars to update. Not needed rn -// for i, executionVariable := range workflowExecution.Workflow.ExecutionVariables { -// if executionVariable.Name == actionResult.Action.ExecutionVariable.Name { -// workflowExecution.Workflow.ExecutionVariables[i] = actionResult.Action.ExecutionVariable -// break -// } -// } + // for i, executionVariable := range workflowExecution.Workflow.ExecutionVariables { + // if executionVariable.Name == actionResult.Action.ExecutionVariable.Name { + // workflowExecution.Workflow.ExecutionVariables[i] = actionResult.Action.ExecutionVariable + // break + // } + // } } else { log.Printf("[DEBUG] NOT updating exec variable %s with new value of length %d. Check previous errors, or if action was successful (success: true)", actionResult.Action.ExecutionVariable.Name, len(actionResult.Result)) @@ -19568,7 +19581,7 @@ func setExecutionVariable(actionResult ActionResult) bool { // Finds execution results and parameters that are too large to manage and reduces them / saves data partly func compressExecution(ctx context.Context, workflowExecution WorkflowExecution, saveLocationInfo string) (WorkflowExecution, bool) { workerCompressExecution := os.Getenv("SHUFFLE_WORKER_COMPRESS") - if project.Environment == "worker" && (len(workerCompressExecution) == 0 || workerCompressExecution == "false"){ + if project.Environment == "worker" && (len(workerCompressExecution) == 0 || workerCompressExecution == "false") { log.Printf("[DEBUG][%s] No need to make this execution any smaller", workflowExecution.ExecutionId) return workflowExecution, false } @@ -21426,7 +21439,7 @@ func HandleDeleteCacheKey(resp http.ResponseWriter, request *http.Request) { DeleteCache(ctx, fmt.Sprintf("%s__%s_%s_100", entity, orgId, cacheData.Category)) DeleteCache(ctx, fmt.Sprintf("%s__%s_%s_1000", entity, orgId, cacheData.Category)) - if debug { + if debug { log.Printf("[DEBUG] Successfully Deleted key '%s' for org %s", cacheKey, orgId) } @@ -21633,7 +21646,7 @@ func HandleDeleteCacheKeyPost(resp http.ResponseWriter, request *http.Request) { Reason: fmt.Sprintf("Key '%s' deleted", tmpData.Key), } - if debug { + if debug { log.Printf("[DEBUG] Successfully Deleted key '%s' for org %s in category '%s'", tmpData.Key, tmpData.OrgId, tmpData.Category) } @@ -22049,7 +22062,6 @@ func HandleSetDatastoreKey(resp http.ResponseWriter, request *http.Request) { return } - body, err := ioutil.ReadAll(request.Body) if err != nil { log.Printf("[WARNING] Failed reading body in set cache: %s", err) @@ -22424,8 +22436,8 @@ func PrepareSingleAction(ctx context.Context, parentRequest *http.Request, user workflowExecution := WorkflowExecution{} if ctx == nil { - ctx = context.Background() - } + ctx = context.Background() + } var action Action err := json.Unmarshal(body, &action) @@ -22600,20 +22612,20 @@ func PrepareSingleAction(ctx context.Context, parentRequest *http.Request, user // Fallback if no group is supplied found := false for _, sensor := range env.SensorHosts { - for _, foundHost := range foundHosts { - if sensor.Hostname == foundHost { + for _, foundHost := range foundHosts { + if sensor.Hostname == foundHost { found = true break } } // Fallback - if found { + if found { parsedEnv = fmt.Sprintf("%s_%s", strings.ToLower(strings.ReplaceAll(strings.ReplaceAll(env.Name, " ", "-"), "_", "-")), env.OrgId) break } } - + continue } @@ -22981,12 +22993,12 @@ func PrepareSingleAction(ctx context.Context, parentRequest *http.Request, user app.ID = action.AppID } - // Prevents overwriting of URL if auth injection is done - shuffleAuthInjected := false + // Prevents overwriting of URL if auth injection is done + shuffleAuthInjected := false // Fallback to inject creds if the user don't have any. This is for internal + // AI oriented APIs only. Check IsShuffleApp() for details - isShuffleApp := IsShuffleApp(app) + isShuffleApp := IsShuffleApp(app) if isShuffleApp && app.Generated && len(workflowExecution.OrgId) > 0 && len(action.AuthenticationId) == 0 && strings.ToLower(app.Name) != "openai" && strings.ToLower(action.Environment) == "cloud" { shuffleAuthInjected = true @@ -23103,9 +23115,9 @@ func PrepareSingleAction(ctx context.Context, parentRequest *http.Request, user action.Parameters[headerIndex].Value = fmt.Sprintf("%s\nOrg-Id: %s", action.Parameters[headerIndex].Value, workflowExecution.OrgId) } - // Custom AI injection when necessary + // Custom AI injection when necessary } else if strings.ToLower(app.Name) == "openai" && len(action.AuthenticationId) == 0 { - shuffleAuthInjected = true + shuffleAuthInjected = true // cloud => only do it on cloud location // This prevents local users from being able to see it if project.Environment != "cloud" || (project.Environment == "cloud" && strings.ToLower(action.Environment) == "cloud") { @@ -23342,8 +23354,8 @@ func PrepareSingleAction(ctx context.Context, parentRequest *http.Request, user // Makes them 'required' to run. Makes it possible to have conditions // for AI Agents in workflows primarily - for _, branch := range oldExec.Workflow.Branches { - if branch.DestinationID != parentActionId { + for _, branch := range oldExec.Workflow.Branches { + if branch.DestinationID != parentActionId { continue } @@ -23553,7 +23565,7 @@ func HandleRetValidation(ctx context.Context, workflowExecution WorkflowExecutio // VERY short sleeptime here on purpose startTime := time.Now().Unix() - maxSeconds := 15 + maxSeconds := 15 if project.Environment != "cloud" { maxSeconds = 180 } @@ -24912,7 +24924,7 @@ func handleOpenIdCloud(resp http.ResponseWriter, request *http.Request) { } // Session management - expiration := time.Now().Add(8 * time.Hour) + expiration := getSessionExpiration() if len(user.Session) == 0 { log.Printf("[INFO] User does NOT have session - creating - (1)") sessionToken := uuid.NewV4().String() @@ -24921,7 +24933,7 @@ func handleOpenIdCloud(resp http.ResponseWriter, request *http.Request) { newCookie.Name = "__session" http.SetCookie(resp, newCookie) - err = SetSession(ctx, user, sessionToken) + err = SetSession(ctx, &user, sessionToken) if err != nil { log.Printf("[WARNING] Error creating session for user: %s", err) resp.WriteHeader(401) @@ -24937,7 +24949,7 @@ func handleOpenIdCloud(resp http.ResponseWriter, request *http.Request) { newCookie.Name = "__session" http.SetCookie(resp, newCookie) - err = SetSession(ctx, user, sessionToken) + err = SetSession(ctx, &user, sessionToken) if err != nil { log.Printf("[WARNING] Error creating session for user: %s", err) resp.WriteHeader(401) @@ -25167,7 +25179,7 @@ func handleOpenIdCloud(resp http.ResponseWriter, request *http.Request) { user.SetSSOInfo(org.Id, orgSSOInfo) // Session management - expiration := time.Now().Add(8 * time.Hour) + expiration := getSessionExpiration() if len(user.Session) == 0 { log.Printf("[INFO] User does NOT have session - creating - (2)") sessionToken := uuid.NewV4().String() @@ -25176,7 +25188,7 @@ func handleOpenIdCloud(resp http.ResponseWriter, request *http.Request) { newCookie.Name = "__session" http.SetCookie(resp, newCookie) - err = SetSession(ctx, user, sessionToken) + err = SetSession(ctx, &user, sessionToken) if err != nil { log.Printf("[WARNING] Error creating session for user: %s", err) resp.WriteHeader(401) @@ -25192,7 +25204,7 @@ func handleOpenIdCloud(resp http.ResponseWriter, request *http.Request) { newCookie.Name = "__session" http.SetCookie(resp, newCookie) - err = SetSession(ctx, user, sessionToken) + err = SetSession(ctx, &user, sessionToken) if err != nil { log.Printf("[WARNING] Error creating session for user: %s", err) resp.WriteHeader(401) @@ -25606,7 +25618,7 @@ func HandleOpenId(resp http.ResponseWriter, request *http.Request) { Role: role, } - expiration := time.Now().Add(8 * time.Hour) + expiration := getSessionExpiration() if len(user.Session) == 0 { log.Printf("[INFO] User does NOT have session - creating - (1)") sessionToken := uuid.NewV4().String() @@ -25617,7 +25629,7 @@ func HandleOpenId(resp http.ResponseWriter, request *http.Request) { newCookie.Name = "__session" http.SetCookie(resp, newCookie) - err = SetSession(ctx, user, sessionToken) + err = SetSession(ctx, &user, sessionToken) if err != nil { log.Printf("[WARNING] Error creating session for user: %s", err) resp.WriteHeader(401) @@ -25635,7 +25647,7 @@ func HandleOpenId(resp http.ResponseWriter, request *http.Request) { newCookie.Name = "__session" http.SetCookie(resp, newCookie) - err = SetSession(ctx, user, sessionToken) + err = SetSession(ctx, &user, sessionToken) if err != nil { log.Printf("[WARNING] Error creating session for user: %s", err) resp.WriteHeader(401) @@ -25781,7 +25793,7 @@ func HandleOpenId(resp http.ResponseWriter, request *http.Request) { Role: role, } - expiration := time.Now().Add(8 * time.Hour) + expiration := getSessionExpiration() if len(user.Session) == 0 { log.Printf("[INFO] User does NOT have session - creating - (2)") sessionToken := uuid.NewV4().String() @@ -25791,7 +25803,7 @@ func HandleOpenId(resp http.ResponseWriter, request *http.Request) { newCookie.Name = "__session" http.SetCookie(resp, newCookie) - err = SetSession(ctx, user, sessionToken) + err = SetSession(ctx, &user, sessionToken) if err != nil { log.Printf("[WARNING] Error creating session for user: %s", err) resp.WriteHeader(401) @@ -25809,7 +25821,7 @@ func HandleOpenId(resp http.ResponseWriter, request *http.Request) { newCookie.Name = "__session" http.SetCookie(resp, newCookie) - err = SetSession(ctx, user, sessionToken) + err = SetSession(ctx, &user, sessionToken) if err != nil { log.Printf("[WARNING] Error creating session for user: %s", err) resp.WriteHeader(401) @@ -25963,7 +25975,7 @@ func HandleOpenId(resp http.ResponseWriter, request *http.Request) { newUser.Id = ID.String() newUser.VerificationToken = verifyToken.String() - expiration := time.Now().Add(8 * time.Hour) + expiration := getSessionExpiration() //if len(user.Session) == 0 { log.Printf("[INFO] User does NOT have session - creating") sessionToken := uuid.NewV4().String() @@ -25974,7 +25986,7 @@ func HandleOpenId(resp http.ResponseWriter, request *http.Request) { newCookie.Name = "__session" http.SetCookie(resp, newCookie) - err = SetSession(ctx, *newUser, sessionToken) + err = SetSession(ctx, newUser, sessionToken) if err != nil { log.Printf("[WARNING] Error creating session for user: %s", err) resp.WriteHeader(401) @@ -26421,7 +26433,7 @@ func PrepareWorkflowExecution(ctx context.Context, workflow Workflow, request *h var execution ExecutionRequest err = json.Unmarshal(body, &execution) if err != nil { - if debug { + if debug { log.Printf("[DEBUG] JSON parsing problem in run workflow: %s", err) } @@ -26435,7 +26447,7 @@ func PrepareWorkflowExecution(ctx context.Context, workflow Workflow, request *h // Ensuring it works even if startpoint isn't defined if execution.Start == "" && len(body) > 0 && len(execution.ExecutionSource) == 0 && len(execution.ExecutionArgument) == 0 { // Check if "execution_argument" in body - if debug { + if debug { log.Printf("[DEBUG] Fallback to full body usage for exec arg") } @@ -26448,7 +26460,7 @@ func PrepareWorkflowExecution(ctx context.Context, workflow Workflow, request *h workflowExecution.ExecutionArgument = execution.ExecutionArgument } - //if debug { + //if debug { // log.Printf("\n\n\n\n\n[DEBUG] INPUT BODY: %s \n\n\n\n\n", string(body)) //} @@ -27055,8 +27067,8 @@ func PrepareWorkflowExecution(ctx context.Context, workflow Workflow, request *h // Parse response to get execution ID and update result var subflowResp struct { - Success bool `json:"success"` - ExecutionID string `json:"execution_id"` + Success bool `json:"success"` + ExecutionID string `json:"execution_id"` Authorization string `json:"authorization"` } if jsonErr := json.Unmarshal(respBody, &subflowResp); jsonErr == nil && len(subflowResp.ExecutionID) > 0 { @@ -27084,17 +27096,17 @@ func PrepareWorkflowExecution(ctx context.Context, workflow Workflow, request *h } // Update result with decline subflow info - updatedResult, marshalErr := json.Marshal(userinputResp) - if marshalErr == nil { - result.Result = string(updatedResult) - for newresIndex, newres := range oldExecution.Results { - if newres.Action.ID == result.Action.ID { - oldExecution.Results[newresIndex] = result - break + updatedResult, marshalErr := json.Marshal(userinputResp) + if marshalErr == nil { + result.Result = string(updatedResult) + for newresIndex, newres := range oldExecution.Results { + if newres.Action.ID == result.Action.ID { + oldExecution.Results[newresIndex] = result + break + } } } } - } log.Printf("[INFO][%s] Decline subflow execution: %s, URL: %s", oldExecution.ExecutionId, subflowResp.ExecutionID, userinputResp.DeclineSubflowURL) } @@ -28216,7 +28228,7 @@ func PrepareWorkflowExecution(ctx context.Context, workflow Workflow, request *h workflowExecution.Workflow.Actions[actionIndex].Environment = "Cloud" cloudExec = true } else { - if project.Environment == "cloud" { + if project.Environment == "cloud" { action.Environment = "Cloud" workflowExecution.Workflow.Actions[actionIndex].Environment = "Cloud" cloudExec = true @@ -31386,7 +31398,7 @@ func GetPriorities(ctx context.Context, user User, org *Org) ([]Priority, error) org, updated = AddPriority(*org, Priority{ Name: fmt.Sprintf("Try Shuffle Security"), - Description: fmt.Sprintf("Automatically handle alerts and vulnerabilities!"), + Description: fmt.Sprintf("Automatically handle alerts and vulnerabilities!"), Type: "security", Active: true, URL: fmt.Sprintf("https://security.shuffler.io"), @@ -37064,8 +37076,8 @@ func getPrioritisedAppActions(ctx context.Context, inputApp string, maxAmount in if !found { returnActions = append(returnActions, action) } else { - if debug { - log.Printf("[DEBUG] NOT adding priority; %#v", action.Name) + if debug { + log.Printf("[DEBUG] NOT adding priority; %#v", action.Name) } } } @@ -37656,22 +37668,22 @@ func collect() ([]ProcessInfo, error) { out := make([]ProcessInfo, 0, len(procs)) for _, p := range procs { ppid, err := p.Ppid() - if err != nil { + if err != nil { ppid = 0 } - tty, err := p.Terminal() // "" if no controlling terminal - if err != nil { + tty, err := p.Terminal() // "" if no controlling terminal + if err != nil { tty = "" } - cmd, err := p.Name() // argv[0] basename - if err != nil { + cmd, err := p.Name() // argv[0] basename + if err != nil { cmd = "" } user, err := p.Username() - if err != nil { + if err != nil { user = "" } @@ -37688,35 +37700,34 @@ func collect() ([]ProcessInfo, error) { args = scrubArgs(args) createdAt, err := p.CreateTime() - if err != nil { + if err != nil { createdAt = 0 } out = append(out, ProcessInfo{ - PID: p.Pid, - PPID: ppid, - TTY: tty, + PID: p.Pid, + PPID: ppid, + TTY: tty, CommandLine: cmd, - User: user, - - Args: args, + User: user, + + Args: args, CreationTime: createdAt, - ExePath: exePath, + ExePath: exePath, // Hash the binary on disk. Note: this is the file at rest, not the // in-memory image — a binary replaced after launch won't be caught here. - SHA256: cachedHashFile(exePath), + SHA256: cachedHashFile(exePath), }) } - if debug { + if debug { log.Printf("[INFO] Found %d processes", len(out)) } return out, nil } - // ListProcesses returns all running processes. // On macOS this calls sysctl kern.proc under the hood. // On Linux this reads /proc. @@ -37948,8 +37959,8 @@ func GetWorkflowAppActions(resp http.ResponseWriter, request *http.Request) { } params = append(params, ActionParameter{ - Name: param.Name, - Required: param.Required, + Name: param.Name, + Required: param.Required, }) } @@ -38060,7 +38071,7 @@ func GetWorkflowMinimal(resp http.ResponseWriter, request *http.Request) { // Permission check: user owns it OR user is in same org if user.Id != workflow.Owner { if workflow.OrgId != user.ActiveOrg.Id { - log.Printf("[WARNING] User %s (%s) unauthorized to view workflow %s (owner: %s, org: %s)", + log.Printf("[WARNING] User %s (%s) unauthorized to view workflow %s (owner: %s, org: %s)", user.Username, user.Id, workflowId, workflow.Owner, workflow.OrgId) resp.WriteHeader(403) resp.Write([]byte(`{"success": false, "reason": "Unauthorized"}`)) @@ -38288,7 +38299,7 @@ CRITICAL RULES FOR THE AGENT backendUrl = fmt.Sprintf("http://localhost:%s", port) } } - + agentReq, err := http.NewRequest("POST", fmt.Sprintf("%s/api/v1/agent", backendUrl), strings.NewReader(string(mcpBody))) if err != nil { log.Printf("[ERROR] Failed creating agent request in AgentWorkflowEditor: %s", err) @@ -38352,7 +38363,6 @@ func createCondition(sourceVal, conditionVal, destVal string) Condition { } } - func findAppByID(ctx context.Context, appID string, user User) (*WorkflowApp, error) { if len(appID) == 0 { return nil, fmt.Errorf("app_id is required") @@ -38368,7 +38378,6 @@ func findAppByID(ctx context.Context, appID string, user User) (*WorkflowApp, er return app, err } - func enrichActionFromApp(ctx context.Context, minAct *MinimalAction, realApp *WorkflowApp, environment string) (Action, error) { if len(realApp.Actions) == 0 { return Action{}, fmt.Errorf("app %s has no actions defined", realApp.Name) @@ -38891,12 +38900,11 @@ func HandleAgentWorkflowSave(resp http.ResponseWriter, request *http.Request) { // authHeader := request.Header.Get("Authorization") // go broadcastBatchToStream(workflow, setOpsReq.Operations, tempIDMap, "agent", "Agent", authHeader) - if debug{ + if debug { log.Printf("[INFO] Applied %d operations to workflow %s for user %s", len(setOpsReq.Operations), workflowID, user.Username) } } - func applyWorkflowOperationWithMapping(ctx context.Context, user User, wf *Workflow, op *WorkflowOperation, tempIDMap map[string]string) error { switch op.Op { // ====== NODE OPERATIONS ====== @@ -38955,7 +38963,6 @@ func findNodePosition(wf *Workflow, nodeID string) (string, int, error) { return "", -1, fmt.Errorf("node %s not found", nodeID) } - func opAddNodeWithMapping(ctx context.Context, user User, wf *Workflow, op *WorkflowOperation, tempIDMap map[string]string) error { err := opAddNode(ctx, user, wf, op) if err != nil { @@ -38997,7 +39004,7 @@ func opAddNode(ctx context.Context, user User, wf *Workflow, op *WorkflowOperati if err != nil { return fmt.Errorf("failed to enrich action: %w", err) } - // Commented out parameter validation to allow agents to add new parameters dynamically + // Commented out parameter validation to allow agents to add new parameters dynamically // for _, param := range newAction.Parameters { // if param.Required && param.Value == "" { // return fmt.Errorf("required parameter '%s' not provided for action %s", param.Name, realApp.Name) @@ -39241,7 +39248,7 @@ func opDeleteNode(wf *Workflow, op *WorkflowOperation) error { if debug { log.Printf("[DEBUG] delete_node(action): action %s not found, already removed - skipping", op.ID) } - + return nil } @@ -39285,7 +39292,6 @@ func opDeleteNode(wf *Workflow, op *WorkflowOperation) error { return nil } - func opAddBranchWithMapping(wf *Workflow, op *WorkflowOperation, tempIDMap map[string]string) error { var branchData struct { SourceID string `json:"source_id"` @@ -39400,11 +39406,10 @@ func opDeleteBranch(wf *Workflow, op *WorkflowOperation) error { if debug { log.Printf("[DEBUG] delete_branch: branch %s not found, already removed (likely cascade from delete_node) - skipping", op.ID) } - + return nil } - func opAddCondition(wf *Workflow, op *WorkflowOperation) error { var condData struct { Source string `json:"source"` @@ -39474,7 +39479,6 @@ func opDeleteCondition(wf *Workflow, op *WorkflowOperation) error { return nil } - func findActionIndexByID(wf *Workflow, id string) int { for i, act := range wf.Actions { if act.ID == id { diff --git a/structs.go b/structs.go index 78291fbf..f11f290a 100755 --- a/structs.go +++ b/structs.go @@ -691,6 +691,8 @@ type User struct { SessionLogin bool `datastore:"session_login" json:"session_login"` // Whether it's a login with session or API (used to verify access) ValidatedSessionOrgs []string `datastore:"validated_session_orgs" json:"validated_session_orgs"` // Orgs that have been used in the current session for the user UsersLastSession string `datastore:"users_last_session" json:"users_last_session"` + SessionCreatedAt int64 `datastore:"session_created_at,noindex" json:"session_created_at,omitempty"` + SessionLastActivityAt int64 `datastore:"session_last_activity_at,noindex" json:"session_last_activity_at,omitempty"` Theme string `datastore:"theme" json:"theme"` PublicProfile PublicProfile `datastore:"public_profile" json:"public_profile"` @@ -727,6 +729,8 @@ type Session struct { Id string `datastore:"Id,noindex"` UserId string `datastore:"user_id,noindex"` Session string `datastore:"session,noindex"` + SessionCreatedAt int64 `datastore:"session_created_at,noindex" json:"session_created_at,omitempty"` + SessionLastActivityAt int64 `datastore:"session_last_activity_at,noindex" json:"session_last_activity_at,omitempty"` } type Contact struct { @@ -3088,28 +3092,27 @@ type Tutorial struct { } type HandleInfo struct { - Success bool `json:"success"` - Admin string `json:"admin"` - Username string `json:"username"` - PublicUsername string `json:"public_username"` - Name string `json:"name"` - ActiveApps []string `json:"active_apps"` - Id string `json:"id"` - Avatar string `json:"avatar"` - Orgs []OrgMini `json:"orgs"` - ActiveOrg OrgMini `json:"active_org"` - EthInfo EthInfo `json:"eth_info,omitempty"` - ChatDisabled bool `json:"chat_disabled"` - Interests []Priority `json:"interests"` - Priorities []Priority `json:"priorities"` - Cookies []SessionCookie `json:"cookies"` - AppExecutionsLimit int64 `json:"app_execution_limit"` - AppExecutionsSuborgs int64 `json:"app_executions_suborgs"` - AppExecutionsUsage int64 `json:"app_execution_usage"` - RegionUrl string `json:"region_url"` - Support bool `json:"support"` - Tutorials []Tutorial `json:"tutorials"` - OrgStatus []string `json:"org_status"` + Success bool `json:"success"` + Admin string `json:"admin"` + Username string `json:"username"` + PublicUsername string `json:"public_username"` + Name string `json:"name"` + ActiveApps []string `json:"active_apps"` + Id string `json:"id"` + Avatar string `json:"avatar"` + Orgs []OrgMini `json:"orgs"` + ActiveOrg OrgMini `json:"active_org"` + EthInfo EthInfo `json:"eth_info,omitempty"` + ChatDisabled bool `json:"chat_disabled"` + Interests []Priority `json:"interests"` + Priorities []Priority `json:"priorities"` + AppExecutionsLimit int64 `json:"app_execution_limit"` + AppExecutionsSuborgs int64 `json:"app_executions_suborgs"` + AppExecutionsUsage int64 `json:"app_execution_usage"` + RegionUrl string `json:"region_url"` + Support bool `json:"support"` + Tutorials []Tutorial `json:"tutorials"` + OrgStatus []string `json:"org_status"` HasCardAvailable bool `json:"has_card_available,omitempty"` ActivatedPayasyougo bool `json:"activated_pay_as_you_go,omitempty"` @@ -4936,11 +4939,11 @@ type MinimalConditionParam struct { // MinimalWorkflow - minimal workflow structure with node positions and connections type MinimalWorkflow struct { - Actions []MinimalAction `json:"actions"` - Branches []MinimalBranch `json:"branches"` - Triggers []MinimalTrigger `json:"triggers"` - Errors []string `json:"errors,omitempty"` - StartTriggerID string `json:"start_trigger_id,omitempty"` + Actions []MinimalAction `json:"actions"` + Branches []MinimalBranch `json:"branches"` + Triggers []MinimalTrigger `json:"triggers"` + Errors []string `json:"errors,omitempty"` + StartTriggerID string `json:"start_trigger_id,omitempty"` } type NGramItem struct { @@ -5331,28 +5334,28 @@ type AppBuildRequest struct { } type AgentsOpsError struct { - Create string `json:"create"` - Run string `json:"run"` - Delete string `json:"delete"` - RunFinished string `json:"run_finished"` + Create string `json:"create"` + Run string `json:"run"` + Delete string `json:"delete"` + RunFinished string `json:"run_finished"` AgentValidation string `json:"agent_validation"` } type AgentHealth struct { - Create bool `json:"create"` - Run bool `json:"run"` - BackendVersion string `json:"backend_version"` - RunFinished bool `json:"run_finished"` - ExecutionTook float64 `json:"execution_took"` - RunStatus string `json:"run_status"` - Delete bool `json:"delete"` - ExecutionId string `json:"execution_id"` - WorkflowId string `json:"workflow_id"` - AgentNodeId string `json:"agent_node_id"` - AgentStatus string `json:"agent_status"` // Status of the agent itself (RUNNING, FINISHED, ABORTED) - AgentDecisionCount int `json:"agent_decision_count"` // Number of decisions made by the agent - LLMCallSuccess bool `json:"llm_call_success"` // Whether the LLM call succeeded - Error AgentsOpsError `json:"error"` + Create bool `json:"create"` + Run bool `json:"run"` + BackendVersion string `json:"backend_version"` + RunFinished bool `json:"run_finished"` + ExecutionTook float64 `json:"execution_took"` + RunStatus string `json:"run_status"` + Delete bool `json:"delete"` + ExecutionId string `json:"execution_id"` + WorkflowId string `json:"workflow_id"` + AgentNodeId string `json:"agent_node_id"` + AgentStatus string `json:"agent_status"` // Status of the agent itself (RUNNING, FINISHED, ABORTED) + AgentDecisionCount int `json:"agent_decision_count"` // Number of decisions made by the agent + LLMCallSuccess bool `json:"llm_call_success"` // Whether the LLM call succeeded + Error AgentsOpsError `json:"error"` } type WttrResponse struct { @@ -5683,10 +5686,11 @@ type DisplaySize struct { OffsetX int `json:"offset_x,omitempty"` OffsetY int `json:"offset_y,omitempty"` } + // Added remote control capabilities for windows -type RemoteControl struct{ - Op string `json:"op"` - Params map[string]any `json:"params"` +type RemoteControl struct { + Op string `json:"op"` + Params map[string]any `json:"params"` } type RemoteControlActionBatch struct { @@ -5709,19 +5713,18 @@ type ActionParameter struct { // ActionSummary - minimal action info for AI agents type ActionSummary struct { - Name string `json:"name"` - Description string `json:"description"` - Parameters []ActionParameter `json:"parameters"` + Name string `json:"name"` + Description string `json:"description"` + Parameters []ActionParameter `json:"parameters"` } // AppActionResponse - actions grouped by app type AppActionResponse struct { - AppName string `json:"app_name"` - AppID string `json:"app_id"` - Actions []ActionSummary `json:"actions"` + AppName string `json:"app_name"` + AppID string `json:"app_id"` + Actions []ActionSummary `json:"actions"` } - // WorkflowOperation represents a single modification operation type WorkflowOperation struct { Op string `json:"op"` // "add_node", "edit_node", "move_node", "delete_node", "add_branch", "edit_branch", "delete_branch", "add_condition", "edit_condition", "delete_condition"