[client] Add support to write connectors bundle to files and/or to queue (#603)

Author: richard-julien

pycti/api/opencti_api_client.py

@@ -108,13 +108,15 @@ def __init__(
         ssl_verify=False,
         proxies=None,
         json_logging=False,
+        bundle_send_to_queue=True,
         cert=None,
         auth=None,
         perform_health_check=True,
     ):
         """Constructor method"""
 
         # Check configuration
+        self.bundle_send_to_queue = bundle_send_to_queue
         self.ssl_verify = ssl_verify
         self.cert = cert
         self.proxies = proxies

pycti/api/opencti_api_work.py

@@ -9,28 +9,34 @@ def __init__(self, api):
         self.api = api
 
     def to_received(self, work_id: str, message: str):
-        self.api.app_logger.info("Reporting work update_received", {"work_id": work_id})
-        query = """
-            mutation workToReceived($id: ID!, $message: String) {
-                workEdit(id: $id) {
-                    toReceived (message: $message)
+        if self.api.bundle_send_to_queue:
+            self.api.app_logger.info(
+                "Reporting work update_received", {"work_id": work_id}
+            )
+            query = """
+                mutation workToReceived($id: ID!, $message: String) {
+                    workEdit(id: $id) {
+                        toReceived (message: $message)
+                    }
                 }
-            }
-           """
-        self.api.query(query, {"id": work_id, "message": message})
+               """
+            self.api.query(query, {"id": work_id, "message": message})
 
     def to_processed(self, work_id: str, message: str, in_error: bool = False):
-        self.api.app_logger.info(
-            "Reporting work update_processed", {"work_id": work_id}
-        )
-        query = """
-            mutation workToProcessed($id: ID!, $message: String, $inError: Boolean) {
-                workEdit(id: $id) {
-                    toProcessed (message: $message, inError: $inError)
+        if self.api.bundle_send_to_queue:
+            self.api.app_logger.info(
+                "Reporting work update_processed", {"work_id": work_id}
+            )
+            query = """
+                mutation workToProcessed($id: ID!, $message: String, $inError: Boolean) {
+                    workEdit(id: $id) {
+                        toProcessed (message: $message, inError: $inError)
+                    }
                 }
-            }
-           """
-        self.api.query(query, {"id": work_id, "message": message, "inError": in_error})
+               """
+            self.api.query(
+                query, {"id": work_id, "message": message, "inError": in_error}
+            )
 
     def ping(self, work_id: str):
         self.api.app_logger.info("Ping work", {"work_id": work_id})
@@ -44,49 +50,52 @@ def ping(self, work_id: str):
         self.api.query(query, {"id": work_id})
 
     def report_expectation(self, work_id: str, error):
-        self.api.app_logger.info("Report expectation", {"work_id": work_id})
-        query = """
-            mutation reportExpectation($id: ID!, $error: WorkErrorInput) {
-                workEdit(id: $id) {
-                    reportExpectation(error: $error)
+        if self.api.bundle_send_to_queue:
+            self.api.app_logger.info("Report expectation", {"work_id": work_id})
+            query = """
+                mutation reportExpectation($id: ID!, $error: WorkErrorInput) {
+                    workEdit(id: $id) {
+                        reportExpectation(error: $error)
+                    }
                 }
-            }
-           """
-        try:
-            self.api.query(query, {"id": work_id, "error": error})
-        except:
-            self.api.app_logger.error("Cannot report expectation")
+               """
+            try:
+                self.api.query(query, {"id": work_id, "error": error})
+            except:
+                self.api.app_logger.error("Cannot report expectation")
 
     def add_expectations(self, work_id: str, expectations: int):
-        self.api.app_logger.info(
-            "Update action expectations",
-            {"work_id": work_id, "expectations": expectations},
-        )
-        query = """
-            mutation addExpectations($id: ID!, $expectations: Int) {
-                workEdit(id: $id) {
-                    addExpectations(expectations: $expectations)
+        if self.api.bundle_send_to_queue:
+            self.api.app_logger.info(
+                "Update action expectations",
+                {"work_id": work_id, "expectations": expectations},
+            )
+            query = """
+                mutation addExpectations($id: ID!, $expectations: Int) {
+                    workEdit(id: $id) {
+                        addExpectations(expectations: $expectations)
+                    }
                 }
-            }
-           """
-        try:
-            self.api.query(query, {"id": work_id, "expectations": expectations})
-        except:
-            self.api.app_logger.error("Cannot report expectation")
+               """
+            try:
+                self.api.query(query, {"id": work_id, "expectations": expectations})
+            except:
+                self.api.app_logger.error("Cannot report expectation")
 
     def initiate_work(self, connector_id: str, friendly_name: str) -> str:
-        self.api.app_logger.info("Initiate work", {"connector_id": connector_id})
-        query = """
-            mutation workAdd($connectorId: String!, $friendlyName: String) {
-                workAdd(connectorId: $connectorId, friendlyName: $friendlyName) {
-                  id
+        if self.api.bundle_send_to_queue:
+            self.api.app_logger.info("Initiate work", {"connector_id": connector_id})
+            query = """
+                mutation workAdd($connectorId: String!, $friendlyName: String) {
+                    workAdd(connectorId: $connectorId, friendlyName: $friendlyName) {
+                      id
+                    }
                 }
-            }
-           """
-        work = self.api.query(
-            query, {"connectorId": connector_id, "friendlyName": friendly_name}
-        )
-        return work["data"]["workAdd"]["id"]
+               """
+            work = self.api.query(
+                query, {"connectorId": connector_id, "friendlyName": friendly_name}
+            )
+            return work["data"]["workAdd"]["id"]
 
     def delete_work(self, work_id: str):
         query = """