Added a dedup+merge action that is easy to use

shuffle-tools/1.2.0/api.yaml

@@ -35,6 +35,55 @@ actions:
         example: print("hello world")
         schema:
           type: string
+
+  - name: dedup_and_merge 
+    description: Merges data from multiple workflows within a set timeframe. Returns action as SKIPPED if the data is a duplicate. Returns with a list of all data if the data at the end
+    parameters:
+      - name: key
+        description: The key to use for deduplication
+        required: true
+        multiline: false
+        example: "ticketname+username"
+        schema:
+          type: string
+      - name: value 
+        description: The full value of the item
+        required: true
+        multiline: true 
+        example: "1208301599081"
+        schema:
+          type: string
+      - name: timeout 
+        description: The timeout before returning
+        required: true
+        options:
+          - 2
+          - 3
+          - 4
+          - 5
+          - 6
+          - 7
+          - 8
+          - 9
+          - 10
+          - 15
+          - 20
+          - 25 
+        multiline: false
+        example: "1"
+        schema:
+          type: string
+      - name: set_skipped 
+        description: Whether to set the action SKIPPED or not IF it matches another workflow in the same timeframe
+        required: true
+        options:
+          - true
+          - false
+        multiline: false
+        example: "true"
+        schema:
+          type: string
+
   - name: check_cache_contains 
     description: Checks Shuffle cache whether a user-provided key contains a value. Returns ALL the values previously appended.
     parameters:

shuffle-tools/1.2.0/src/app.py

@@ -221,6 +221,79 @@ def send_email_shuffle(self, apikey, recipients, subject, body, attachments=""):
     def repeat_back_to_me(self, call):
         return call
 
+    def dedup_and_merge(self, key, value, timeout, set_skipped=True):
+        timeout = int(timeout)
+        key = str(key)
+
+        set_skipped = True
+        if str(set_skipped).lower() == "false":
+            set_skipped = False
+        else:
+            set_skipped = True
+
+        cachekey = "dedup-%s" % (key)
+        response = {
+            "success": False,
+            "datastore_key": cachekey,
+            "info": "All keys from the last %d seconds with the key '%s' have been merged. The result was set to SKIPPED in all other actions." % (timeout, key),
+            "timeout": timeout,
+            "original_value": value,
+            "all_values": [],
+        }
+
+        found_cache = self.get_cache(cachekey)
+
+        if found_cache["success"] == True and len(found_cache["value"]) > 0:
+            if "value" in found_cache:
+                if not str(found_cache["value"]).startswith("["):
+                    found_cache["value"] = [found_cache["value"]]
+                else:
+                    try:
+                        found_cache["value"] = json.loads(found_cache["value"])
+                    except Exception as e:
+                        self.logger.info("[ERROR] Failed parsing JSON: %s" % e)
+            else:
+                found_cache["value"] = []
+
+            found_cache["value"].append(value)
+            if "created" in found_cache:
+                if found_cache["created"] + timeout + 3 < time.time():
+                    set_skipped = False 
+                    response["success"] = True
+                    response["all_values"] = found_cache["value"]
+
+                    self.delete_cache(cachekey)
+
+                    return json.dumps(response)
+                else:
+                    self.logger.info("Dedup-key is already handled in another workflow with timeout %d" % timeout)
+
+            self.set_cache(cachekey, json.dumps(found_cache["value"]))
+            if set_skipped == True:
+                self.action_result["status"] = "SKIPPED"
+                self.action_result["result"] = json.dumps({
+                    "status": False,
+                    "reason": "Dedup-key is already handled in another workflow with timeout %d" % timeout,
+                })
+
+                self.send_result(self.action_result, {"Authorization": "Bearer %s" % self.authorization}, "/api/v1/streams")
+
+            return found_cache
+
+        parsedvalue = [value]
+        resp = self.set_cache(cachekey, json.dumps(parsedvalue))
+
+        self.logger.info("Sleeping for %d seconds while waiting for cache to fill up elsewhere" % timeout)
+        time.sleep(timeout)
+        found_cache = self.get_cache(cachekey)
+
+        response["success"] = True
+        response["all_values"] = found_cache["value"]
+
+        self.delete_cache(cachekey)
+        return json.dumps(response)
+
+
     # https://github.com/fhightower/ioc-finder
     def parse_file_ioc(self, file_ids, input_type="all"):
         def parse(data):