[client] New specific operations for PIR (opencti #10032)

Co-authored-by: Cathia Archidoit <[email protected]>
Co-authored-by: Souad Hadjiat <[email protected]>

Author: 3 people

pycti/api/opencti_api_client.py

@@ -10,6 +10,7 @@
 
 from pycti import __version__
 from pycti.api.opencti_api_connector import OpenCTIApiConnector
+from pycti.api.opencti_api_pir import OpenCTIApiPir
 from pycti.api.opencti_api_playbook import OpenCTIApiPlaybook
 from pycti.api.opencti_api_work import OpenCTIApiWork
 from pycti.entities.opencti_attack_pattern import AttackPattern
@@ -170,6 +171,7 @@ def __init__(
         self.playbook = OpenCTIApiPlaybook(self)
         self.connector = OpenCTIApiConnector(self)
         self.stix2 = OpenCTIStix2(self)
+        self.pir = OpenCTIApiPir(self)
 
         # Define the entities
         self.vocabulary = Vocabulary(self)

pycti/api/opencti_api_pir.py

@@ -0,0 +1,37 @@
+class OpenCTIApiPir:
+    """OpenCTIApiPir"""
+
+    def __init__(self, api):
+        self.api = api
+
+    def pir_flag_element(self, **kwargs):
+        id = kwargs.get("id", None)
+        input = kwargs.get("input", None)
+        query = """
+            mutation PirFlagElement($id: ID!, $input: PirFlagElementInput!) {
+                pirFlagElement(id: $id, input: $input)
+            }
+           """
+        self.api.query(
+            query,
+            {
+                "id": id,
+                "input": input,
+            },
+        )
+
+    def pir_unflag_element(self, **kwargs):
+        id = kwargs.get("id", None)
+        input = kwargs.get("input", None)
+        query = """
+            mutation PirUnflagElement($id: ID!, $input: PirUnflagElementInput!) {
+                pirUnflagElement(id: $id, input: $input)
+            }
+           """
+        self.api.query(
+            query,
+            {
+                "id": id,
+                "input": input,
+            },
+        )

pycti/utils/opencti_stix2.py

@@ -2484,6 +2484,14 @@ def apply_opencti_operation(self, item, operation):
             self.opencti.stix.merge(id=target_id, object_ids=source_ids)
         elif operation == "patch":
             self.apply_patch(item=item)
+        elif operation == "pir_flag_element":
+            id = item["id"]
+            input = item["input"]
+            self.opencti.pir.pir_flag_element(id=id, input=input)
+        elif operation == "pir_unflag_element":
+            id = item["id"]
+            input = item["input"]
+            self.opencti.pir.pir_unflag_element(id=id, input=input)
         else:
             raise ValueError("Not supported opencti_operation")
 

pycti/utils/opencti_stix2_splitter.py

@@ -19,6 +19,7 @@
     SUPPORTED_STIX_ENTITY_OBJECTS  # entities
     + list(STIX_CYBER_OBSERVABLE_MAPPING.keys())  # observables
     + ["relationship", "sighting"]  # relationships
+    + ["pir"]
 )