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

SDK & ORborus update

Browse Source
This commit is contained in:
Frikky
2024-08-16 16:52:31 +02:00
parent 7419482c45
commit 5aba926ef0
2 changed files with 40 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app_sdk/app_base.py
+10 -7
View File
@@ -685,6 +685,7 @@ class AppBase:
if not "User-Agent" in headers:
headers["User-Agent"] = "Shuffle App"
self.logger.info(f"[DEBUG][{self.current_execution_id}] Starting to send result to {url}")
try:
finished = False
ret = {}
@@ -702,8 +703,8 @@ class AppBase:
proxies=self.proxy_config,
)
#self.logger.info(f"""[DEBUG] Successful result request: Status= {ret.status_code} (break on 200/201) & Action status: {action_result["status"]}. Response= {ret.text}""")
if ret.status_code == 200 or ret.status_code == 201:
self.logger.info(f"[DEBUG][{self.current_execution_id}] Successful send_result request: Status= {ret.status_code} & Action status: {action_result['status']}. Response= {ret.text}")
finished = True
break
else:
@@ -1258,6 +1259,7 @@ class AppBase:
# self.send_result(action_result, headers, stream_path)
# return
self.logger.info(f"[DEBUG][{self.current_execution_id}] Pre function with {len(param_multiplier)} multipliers")
for subparams in param_multiplier:
#self.logger.info(f"SUBPARAMS IN MULTI: {subparams}")
tmp = ""
@@ -1355,6 +1357,8 @@ class AppBase:
#ret = ret[0]
self.logger.info("[DEBUG] DONT make list of 1 into 0!!")
self.logger.info(f"[DEBUG][%s] Done with execution recursion %d times" % (self.current_execution_id, len(param_multiplier)))
#self.logger.info("Return from execution: %s" % ret)
if ret == None:
results.append("")
@@ -1923,14 +1927,14 @@ class AppBase:
"""
Generate strings contained in nested (), indexing i = level
"""
if len(re.findall("\(", string)) == len(re.findall("\)", string)):
if len(re.findall('(', string)) == len(re.findall(')', string)):
LeftRightIndex = [x for x in zip(
[Left.start()+1 for Left in re.finditer('\(', string)],
reversed([Right.start() for Right in re.finditer('\)', string)]))]
[Left.start()+1 for Left in re.finditer('(', string)],
reversed([Right.start() for Right in re.finditer(')', string)]))]
elif len(re.findall("\(", string)) > len(re.findall("\)", string)):
elif len(re.findall('(', string)) > len(re.findall(')', string)):
return parse_nested_param(string + ')', level)
elif len(re.findall("\(", string)) < len(re.findall("\)", string)):
elif len(re.findall('(', string)) < len(re.findall('"', string)):
return parse_nested_param('(' + string, level)
else:
return 'Failed to parse params'
@@ -3388,7 +3392,6 @@ class AppBase:
elif callable(func):
try:
if len(action["parameters"]) < 1:
#result = await func()
result = func()
else:
# Potentially parse JSON here
functions/onprem/orborus/orborus.go
+30 -17
View File
@@ -101,9 +101,10 @@ var swarmConfig = os.Getenv("SHUFFLE_SWARM_CONFIG")
var swarmNetworkName = os.Getenv("SHUFFLE_SWARM_NETWORK_NAME")
var orborusLabel = os.Getenv("SHUFFLE_ORBORUS_LABEL")
var memcached = os.Getenv("SHUFFLE_MEMCACHED")
var tenzirUrl = os.Getenv("SHUFFLE_TENZIR_URL")
var apiKey = os.Getenv("AUTH_FOR_ORBORUS")
var pipelineUrl = os.Getenv("SHUFFLE_PIPELINE_URL")
var executionIds = []string{}
var namespacemade = false // For K8s
@@ -1807,9 +1808,22 @@ func main() {
log.Printf("[WARNING] Defaulting to environment name %s. Set environment variable ENVIRONMENT_NAME to change. This should be the same as in the frontend action.", environment)
}
if tenzirUrl == "" {
tenzirUrl = "http://localhost:5160"
log.Printf("[WARNING] SHUFFLE_TENZIR_URL not set, falling back to default URL: %s",tenzirUrl)
if pipelineUrl == "" {
pipelineUrl = "http://localhost:5160"
// Find the IP in baseUrl. Base format is http://<ip>:<port>
if baseUrl != "" {
urlSplit := strings.Split(baseUrl, "://")
if len(urlSplit) > 1 {
// Find the IP
ipSplit := strings.Split(urlSplit[1], ":")
if len(ipSplit) > 0 {
pipelineUrl = fmt.Sprintf("http://%s:5160", ipSplit[0])
}
}
}
log.Printf("[WARNING] SHUFFLE_PIPELINE_URL not set, falling back to default URL: %s. If BASE_URL is set, we use the external IP for that",pipelineUrl)
}
@@ -2035,7 +2049,7 @@ func main() {
err := deployTenzirNode()
if err != nil{
log.Printf("[ERROR] failed to deploy the pipeline, reason: %s", err)
log.Printf("[ERROR] Failed to deploy CATEGORY UPDATE, reason: %s", err)
}
err = handleFileCategoryChange()
@@ -2049,7 +2063,7 @@ func main() {
fileName := incRequest.ExecutionArgument
err := deployTenzirNode()
if err != nil{
log.Printf("[ERROR] failed to deploy the pipeline, reason: %s", err)
log.Printf("[ERROR] Failed to deploy DISABLE SIGMA FILE, reason: %s", err)
}
err = disableRule(fileName)
@@ -2063,7 +2077,7 @@ func main() {
fileName := incRequest.ExecutionArgument
err := deployTenzirNode()
if err != nil{
log.Printf("[ERROR] failed to deploy the pipeline, reason: %s", err)
log.Printf("[ERROR] Failed to deploy ENABLE SIGMA FILE, reason: %s", err)
}
err = enableRule(fileName)
@@ -2074,10 +2088,9 @@ func main() {
toBeRemoved.Data = append(toBeRemoved.Data, incRequest)
} else if incRequest.Type == "DISABLE_SIGMA_FOLDER" {
err := deployTenzirNode()
if err != nil{
log.Printf("[ERROR] failed to deploy the pipeline, reason: %s", err)
log.Printf("[ERROR] Failed to deploy DISABLE SIGMA FOLDER, reason: %s", err)
}
err = removeAllFiles()
@@ -2090,7 +2103,7 @@ func main() {
log.Printf("[INFO] Got job to start tenzir")
err := deployTenzirNode()
if err != nil{
log.Printf("[ERROR] failed to deploy the pipeline, reason: %s", err)
log.Printf("[ERROR] Failed to deploy the pipeline, reason: %s", err)
}
toBeRemoved.Data = append(toBeRemoved.Data, incRequest)
@@ -2446,7 +2459,7 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
err := deployTenzirNode()
if err != nil {
log.Printf("[ERROR] failed to deploy the pipeline, reason: %s", err)
log.Printf("[ERROR] Failed to deploy the pipeline, reason: %s", err)
return err
}
@@ -2557,7 +2570,7 @@ func deployTenzirNode() error {
// Check if image exists
_, _, err := dockercli.ImageInspectWithRaw(ctx, imageName)
if dockerclient.IsErrNotFound(err) {
log.Printf("[DEBUG] pulling image %s", imageName)
log.Printf("[DEBUG] Pulling image %s", imageName)
pullOptions := image.PullOptions{}
out, err := dockercli.ImagePull(ctx, imageName, pullOptions)
if err != nil {
@@ -2713,7 +2726,7 @@ func createNetworkIfNotExists(ctx context.Context, networkName, subnet, gateway
func checkTenzirNode() error {
retries := 5
retryInterval := 3 * time.Second
url := fmt.Sprintf("%s/api/v0/ping",tenzirUrl)
url := fmt.Sprintf("%s/api/v0/ping",pipelineUrl)
forwardMethod := "POST"
client := http.Client{}
@@ -2739,7 +2752,7 @@ func createPipeline(command, identifier string) (string, error) {
toBeDeleted := false
pipelineId, err := searchPipeline(identifier)
url := fmt.Sprintf("%s/api/v0/pipeline/create", tenzirUrl)
url := fmt.Sprintf("%s/api/v0/pipeline/create", pipelineUrl)
forwardMethod := "POST"
if err != nil {
@@ -2849,7 +2862,7 @@ func createPipeline(command, identifier string) (string, error) {
func updatePipelineState(command, pipelineId, action string) (string, error) {
url := fmt.Sprintf("%s/api/v0/pipeline/update", tenzirUrl)
url := fmt.Sprintf("%s/api/v0/pipeline/update", pipelineUrl)
forwardMethod := "POST"
requestBody := map[string]interface{}{
@@ -2919,7 +2932,7 @@ func deletePipeline(pipelineId string) error {
"id": pipelineId,
}
url := fmt.Sprintf("%s/api/v0/pipeline/delete", tenzirUrl)
url := fmt.Sprintf("%s/api/v0/pipeline/delete", pipelineUrl)
forwardMethod := "POST"
requestBodyJSON, err := json.Marshal(requestBody)
@@ -2967,7 +2980,7 @@ func searchPipeline(identifier string) (string, error) {
var reqBody []byte
url := fmt.Sprintf("%s/api/v0/pipeline/list", tenzirUrl)
url := fmt.Sprintf("%s/api/v0/pipeline/list", pipelineUrl)
resp, err := http.Post(url, "application/json", bytes.NewBuffer(reqBody))
if err != nil {