[client] Allow consume remote custom live stream

Author: Samuel Hassine

pycti/connector/opencti_connector_helper.py

@@ -242,7 +242,7 @@ def stop(self) -> None:
 
 class ListenStream(threading.Thread):
     def __init__(
-        self, helper, callback, url, token, verify_ssl, start_timestamp
+        self, helper, callback, url, token, verify_ssl, start_timestamp, live_stream_id
     ) -> None:
         threading.Thread.__init__(self)
         self.helper = helper
@@ -251,6 +251,7 @@ def __init__(
         self.token = token
         self.verify_ssl = verify_ssl
         self.start_timestamp = start_timestamp
+        self.live_stream_id = live_stream_id
         self.exit = False
 
     def run(self) -> None:  # pylint: disable=too-many-branches
@@ -266,13 +267,17 @@ def run(self) -> None:  # pylint: disable=too-many-branches
         # If URL and token are provided, likely consuming a remote stream
         if self.url is not None and self.token is not None:
             # If a live stream ID, appending the URL
-            live_stream_uri = (
-                f"/{self.helper.connect_live_stream_id}"
-                if self.helper.connect_live_stream_id is not None
-                else ""
-            )
+            if self.live_stream_id is not None:
+                live_stream_uri = f"/{self.live_stream_id}"
+            elif self.helper.connect_live_stream_id is not None:
+                live_stream_uri = f"/{self.helper.connect_live_stream_id}"
+            else:
+                live_stream_uri = ""
             # Live stream "from" should be empty if start from the beginning
-            if self.helper.connect_live_stream_id is not None:
+            if (
+                self.live_stream_id is not None
+                or self.helper.connect_live_stream_id is not None
+            ):
                 live_stream_from = (
                     f"?from={current_state['connectorLastEventId']}"
                     if current_state["connectorLastEventId"] != "-"
@@ -509,14 +514,21 @@ def listen_stream(
         token=None,
         verify_ssl=None,
         start_timestamp=None,
+        live_stream_id=None,
     ) -> ListenStream:
         """listen for messages and register callback function
 
         :param message_callback: callback function to process messages
         """
 
         self.listen_stream = ListenStream(
-            self, message_callback, url, token, verify_ssl, start_timestamp
+            self,
+            message_callback,
+            url,
+            token,
+            verify_ssl,
+            start_timestamp,
+            live_stream_id,
         )
         self.listen_stream.start()
         return self.listen_stream