[client] Fix synchronization between platforms

Author: Samuel Hassine

pycti/connector/opencti_connector_helper.py

@@ -219,18 +219,23 @@ def run(self):
 
 
 class ListenStream(threading.Thread):
-    def __init__(self, helper, callback, url, token, verify_ssl):
+    def __init__(self, helper, callback, url, token, verify_ssl, start_timestamp):
         threading.Thread.__init__(self)
         self.helper = helper
         self.callback = callback
         self.url = url
         self.token = token
         self.verify_ssl = verify_ssl
+        self.start_timestamp = start_timestamp
 
     def run(self):
         current_state = self.helper.get_state()
         if current_state is None:
-            current_state = {"connectorLastEventId": "-"}
+            current_state = {
+                "connectorLastEventId": str(self.start_timestamp) + "-0"
+                if self.start_timestamp is not None and len(self.start_timestamp) > 0
+                else "-"
+            }
             self.helper.set_state(current_state)
 
         # If URL and token are provided, likely consuming a remote stream
@@ -462,13 +467,16 @@ def listen_stream(
         url=None,
         token=None,
         verify_ssl=None,
+        start_timestamp=None,
     ) -> None:
         """listen for messages and register callback function
 
         :param message_callback: callback function to process messages
         """
 
-        listen_stream = ListenStream(self, message_callback, url, token, verify_ssl)
+        listen_stream = ListenStream(
+            self, message_callback, url, token, verify_ssl, start_timestamp
+        )
         listen_stream.start()
 
     def get_opencti_url(self):
@@ -515,11 +523,12 @@ def send_stix2_bundle(self, bundle, **kwargs) -> list:
         work_id = kwargs.get("work_id", self.work_id)
         entities_types = kwargs.get("entities_types", None)
         update = kwargs.get("update", False)
+        event_version = kwargs.get("event_version", None)
 
         if entities_types is None:
             entities_types = []
         stix2_splitter = OpenCTIStix2Splitter()
-        bundles = stix2_splitter.split_bundle(bundle)
+        bundles = stix2_splitter.split_bundle(bundle, True, event_version)
         if len(bundles) == 0:
             raise ValueError("Nothing to import")
         if work_id is not None:

pycti/entities/opencti_stix_core_relationship.py

@@ -535,6 +535,10 @@ def update_field(self, **kwargs):
         id = kwargs.get("id", None)
         key = kwargs.get("key", None)
         value = kwargs.get("value", None)
+        if isinstance(value, list):
+            value = [str(v) for v in value]
+        else:
+            value = str(value)
         if id is not None and key is not None and value is not None:
             self.opencti.log(
                 "info",

pycti/entities/opencti_stix_cyber_observable.py

@@ -1078,7 +1078,10 @@ def update_field(self, **kwargs):
         key = kwargs.get("key", None)
         value = kwargs.get("value", None)
         operation = kwargs.get("operation", "replace")
-
+        if isinstance(value, list):
+            value = [str(v) for v in value]
+        else:
+            value = str(value)
         if id is not None and key is not None and value is not None:
             self.opencti.log(
                 "info", "Updating Stix-Observable {" + id + "} field {" + key + "}."

pycti/entities/opencti_stix_domain_object.py

@@ -577,7 +577,10 @@ def update_field(self, **kwargs):
         key = kwargs.get("key", None)
         value = kwargs.get("value", None)
         operation = kwargs.get("operation", "replace")
-
+        if isinstance(value, list):
+            value = [str(v) for v in value]
+        else:
+            value = str(value)
         if id is not None and key is not None and value is not None:
             self.opencti.log(
                 "info", "Updating Stix-Domain-Object {" + id + "} field {" + key + "}."

pycti/utils/opencti_stix2.py

@@ -1490,9 +1490,14 @@ def import_bundle(self, stix_bundle, update=False, types=None) -> List:
             raise ValueError("JSON data type is not a STIX2 bundle")
         if "objects" not in stix_bundle or len(stix_bundle["objects"]) == 0:
             raise ValueError("JSON data objects is empty")
+        event_version = (
+            stix_bundle["x_opencti_event_version"]
+            if "x_opencti_event_version" in stix_bundle
+            else None
+        )
 
         stix2_splitter = OpenCTIStix2Splitter()
-        bundles = stix2_splitter.split_bundle(stix_bundle, False)
+        bundles = stix2_splitter.split_bundle(stix_bundle, False, event_version)
         # Import every elements in a specific order
         imported_elements = []
 
@@ -1503,10 +1508,12 @@ def import_bundle(self, stix_bundle, update=False, types=None) -> List:
                     if bundle["x_opencti_event_version"] == "1":
                         if "x_data_update" in item:
                             self.stix2_update.process_update_v1(item)
+                            continue
                     elif bundle["x_opencti_event_version"] == "2":
-                        if "x_opencti_patch":
+                        if "x_opencti_patch" in item:
                             self.stix2_update.process_update_v2(item)
-                elif item["type"] == "relationship":
+                            continue
+                if item["type"] == "relationship":
                     self.import_relationship(item, update, types)
                 elif item["type"] == "sighting":
                     # Resolve the to

pycti/utils/opencti_stix2_splitter.py

@@ -35,7 +35,7 @@ def enlist_element(self, item_id, raw_data):
         self.cache_index[item_id] = item  # Put in cache
         return nb_deps
 
-    def split_bundle(self, bundle, use_json=True) -> list:
+    def split_bundle(self, bundle, use_json=True, event_version=None) -> list:
         """splits a valid stix2 bundle into a list of bundles
         :param bundle: valid stix2 bundle
         :type bundle:
@@ -70,11 +70,11 @@ def by_dep_size(elem):
 
         self.elements.sort(key=by_dep_size)
         for entity in self.elements:
-            bundles.append(self.stix2_create_bundle([entity], use_json))
+            bundles.append(self.stix2_create_bundle([entity], use_json, event_version))
         return bundles
 
     @staticmethod
-    def stix2_create_bundle(items, use_json):
+    def stix2_create_bundle(items, use_json, event_version=None):
         """create a stix2 bundle with items
 
         :param items: valid stix2 items
@@ -91,4 +91,6 @@ def stix2_create_bundle(items, use_json):
             "spec_version": "2.1",
             "objects": items,
         }
+        if event_version is not None:
+            bundle["x_opencti_event_version"] = event_version
         return json.dumps(bundle) if use_json else bundle