[client] handle notification update & delete

Author: JeremyCloarec

pycti/api/opencti_api_client.py

@@ -11,6 +11,7 @@
 from pycti import __version__
 from pycti.api.opencti_api_connector import OpenCTIApiConnector
 from pycti.api.opencti_api_draft import OpenCTIApiDraft
+from pycti.api.opencti_api_notification import OpenCTIApiNotification
 from pycti.api.opencti_api_pir import OpenCTIApiPir
 from pycti.api.opencti_api_playbook import OpenCTIApiPlaybook
 from pycti.api.opencti_api_public_dashboard import OpenCTIApiPublicDashboard
@@ -172,6 +173,7 @@ def __init__(
         self.session = requests.session()
         # Define the dependencies
         self.work = OpenCTIApiWork(self)
+        self.notification = OpenCTIApiNotification(self)
         self.trash = OpenCTIApiTrash(self)
         self.draft = OpenCTIApiDraft(self)
         self.workspace = OpenCTIApiWorkspace(self)

pycti/api/opencti_api_notification.py

@@ -0,0 +1,39 @@
+class OpenCTIApiNotification:
+    """OpenCTIApiJob"""
+
+    def __init__(self, api):
+        self.api = api
+
+    def delete(self, **kwargs):
+        notification_id = kwargs.get("id", None)
+        self.api.app_logger.info(
+            "Deleting notifcation", {"notification_id": notification_id}
+        )
+        query = """
+            mutation notificationDelete($id: ID!) {
+                notificationDelete(id: $id)
+            }
+           """
+        self.api.query(query, {"id": notification_id})
+
+    def update_field(self, **kwargs):
+        notification_id = kwargs.get("id", None)
+        input = kwargs.get("input", None)
+        for input_value in input:
+            if input_value["key"] == "is_read":
+                is_read_value = bool(input_value["value"][0])
+                self.mark_as_read(notification_id, is_read_value)
+
+    def mark_as_read(self, notification_id: str, read: bool):
+        self.api.app_logger.info(
+            "Marking notifcation as read",
+            {"notification_id": notification_id, "read": read},
+        )
+        query = """
+                mutation notificationMarkRead($id: ID!, $read: Boolean!) {
+                    notificationMarkRead(id: $id, read: $read) {
+                        id
+                    }
+                }
+               """
+        self.api.query(query, {"id": notification_id, "read": read})

pycti/utils/opencti_stix2.py

@@ -925,6 +925,7 @@ def get_internal_helper(self):
             "playbook": self.opencti.playbook,
             "workspace": self.opencti.workspace,
             "publicdashboard": self.opencti.public_dashboard,
+            "notification": self.opencti.notification,
         }
 
     def generate_standard_id_from_stix(self, data):
@@ -2490,6 +2491,10 @@ def apply_patch(self, item):
                 self.opencti.indicator.update_field(
                     id=item_id, input=field_patch_without_files
                 )
+            elif item["type"] == "notification":
+                self.opencti.notification.update_field(
+                    id=item_id, input=field_patch_without_files
+                )
             else:
                 self.opencti.stix_domain_object.update_field(
                     id=item_id, input=field_patch_without_files

pycti/utils/opencti_stix2_utils.py

@@ -8,6 +8,7 @@
     "capability",
     "role",
     "settings",
+    "notification",
     "work",
     "trash",
     "draftworkspace",