[client] Introduce new class

Author: richard-julien

pycti/api/opencti_api_client.py

@@ -12,8 +12,10 @@
 from pycti.api.opencti_api_connector import OpenCTIApiConnector
 from pycti.api.opencti_api_draft import OpenCTIApiDraft
 from pycti.api.opencti_api_playbook import OpenCTIApiPlaybook
+from pycti.api.opencti_api_public_dashboard import OpenCTIApiPublicDashboard
 from pycti.api.opencti_api_trash import OpenCTIApiTrash
 from pycti.api.opencti_api_work import OpenCTIApiWork
+from pycti.api.opencti_api_workspace import OpenCTIApiWorkspace
 from pycti.entities.opencti_attack_pattern import AttackPattern
 from pycti.entities.opencti_campaign import Campaign
 from pycti.entities.opencti_capability import Capability
@@ -171,6 +173,8 @@ def __init__(
         self.work = OpenCTIApiWork(self)
         self.trash = OpenCTIApiTrash(self)
         self.draft = OpenCTIApiDraft(self)
+        self.workspace = OpenCTIApiWorkspace(self)
+        self.public_dashboard = OpenCTIApiPublicDashboard(self)
         self.playbook = OpenCTIApiPlaybook(self)
         self.connector = OpenCTIApiConnector(self)
         self.stix2 = OpenCTIStix2(self)

pycti/api/opencti_api_draft.py

@@ -4,7 +4,8 @@ class OpenCTIApiDraft:
     def __init__(self, api):
         self.api = api
 
-    def delete(self, draft_id: str):
+    def delete(self, **kwargs):
+        id = kwargs.get("id", None)
         query = """
             mutation DraftWorkspaceDelete($id: ID!) {
                 draftWorkspaceDelete(id: $id)
@@ -13,6 +14,6 @@ def delete(self, draft_id: str):
         self.api.query(
             query,
             {
-                "id": draft_id,
+                "id": id,
             },
         )

pycti/api/opencti_api_playbook.py

@@ -32,3 +32,17 @@ def playbook_step_execution(self, playbook: dict, bundle: str):
                 "bundle": bundle,
             },
         )
+
+    def delete(self, **kwargs):
+        id = kwargs.get("id", None)
+        query = """
+            mutation PlaybookDelete($id: ID!) {
+                playbookDelete(id: $id)
+            }
+           """
+        self.api.query(
+            query,
+            {
+                "id": id,
+            },
+        )

pycti/api/opencti_api_public_dashboard.py

@@ -0,0 +1,19 @@
+class OpenCTIApiPublicDashboard:
+    """OpenCTIApiPublicDashboard"""
+
+    def __init__(self, api):
+        self.api = api
+
+    def delete(self, **kwargs):
+        id = kwargs.get("id", None)
+        query = """
+            mutation PublicDashboardDelete($id: ID!) {
+                publicDashboardDelete(id: $id)
+            }
+           """
+        self.api.query(
+            query,
+            {
+                "id": id,
+            },
+        )

pycti/api/opencti_api_trash.py

@@ -4,7 +4,7 @@ class OpenCTIApiTrash:
     def __init__(self, api):
         self.api = api
 
-    def delete_operation_restore(self, operation_id: str):
+    def restore(self, operation_id: str):
         query = """
             mutation DeleteOperationRestore($id: ID!) {
                 deleteOperationRestore(id: $id)
@@ -16,3 +16,26 @@ def delete_operation_restore(self, operation_id: str):
                 "id": operation_id,
             },
         )
+
+    def delete(self, **kwargs):
+        """Delete a role given its ID
+
+        :param id: ID for the role on the platform.
+        :type id: str
+        """
+        id = kwargs.get("id", None)
+        if id is None:
+            self.api.admin_logger.error("[opencti_role] Missing parameter: id")
+            return None
+
+        query = """
+            mutation DeleteOperationConfirm($id: ID!) {
+                deleteOperationConfirm(id: $id) {
+            }
+        """
+        self.api.query(
+            query,
+            {
+                "id": id,
+            },
+        )

pycti/api/opencti_api_work.py

@@ -128,6 +128,20 @@ def delete_work(self, work_id: str):
         )
         return work["data"]
 
+    def delete(self, **kwargs):
+        id = kwargs.get("id", None)
+        query = """
+        mutation ConnectorWorksMutation($workId: ID!) {
+            workEdit(id: $workId) {
+                delete
+            }
+        }"""
+        work = self.api.query(
+            query,
+            {"workId": id},
+        )
+        return work["data"]
+
     def wait_for_work_to_finish(self, work_id: str):
         status = ""
         cnt = 0

pycti/api/opencti_api_workspace.py

@@ -0,0 +1,19 @@
+class OpenCTIApiWorkspace:
+    """OpenCTIApiWorkspace"""
+
+    def __init__(self, api):
+        self.api = api
+
+    def delete(self, **kwargs):
+        id = kwargs.get("id", None)
+        query = """
+            mutation WorkspaceDelete($id: ID!) {
+                workspaceDelete(id: $id)
+            }
+           """
+        self.api.query(
+            query,
+            {
+                "id": id,
+            },
+        )

pycti/entities/opencti_group.py

@@ -315,12 +315,16 @@ def create(self, **kwargs) -> Optional[Dict]:
         )
         return self.opencti.process_multiple_fields(result["data"]["groupAdd"])
 
-    def delete(self, id: str):
+    def delete(self, **kwargs):
         """Delete a given group from OpenCTI
 
         :param id: ID of the group to delete.
         :type id: str
         """
+        id = kwargs.get("id", None)
+        if id is None:
+            self.opencti.admin_logger.error("[opencti_user] Missing parameter: id")
+            return None
         self.opencti.admin_logger.info("Deleting group", {"id": id})
         query = """
             mutation GroupDelete($id: ID!) {

pycti/entities/opencti_stix_core_object.py

@@ -1910,4 +1910,4 @@ def remove_from_draft(self, **kwargs):
             self.opencti.query(query, {"id": id})
         else:
             self.opencti.app_logger.error("[stix_core_object] Missing parameters: id")
-            return None
+            return None

pycti/utils/constants.py

@@ -48,7 +48,7 @@ class StixCyberObservableTypes(Enum):
     PERSONA = "Persona"
 
     @classmethod
-    def has_value(cls, value):
+    def has_value(cls, value: str) -> bool:
         lower_attr = list(map(lambda x: x.lower(), cls._value2member_map_))
         return value.lower() in lower_attr