[client] Renaming Threat-Actor into Threat-Actor-Group (#3166)

Co-authored-by: sarahBocognano <[email protected]>
Co-authored-by: Julien Richard <[email protected]>

Author: 3 people

pycti/__init__.py

@@ -45,6 +45,7 @@
 from .entities.opencti_stix_sighting_relationship import StixSightingRelationship
 from .entities.opencti_task import Task
 from .entities.opencti_threat_actor import ThreatActor
+from .entities.opencti_threat_actor_group import ThreatActorGroup
 from .entities.opencti_tool import Tool
 from .entities.opencti_vulnerability import Vulnerability
 from .utils.constants import (
@@ -111,6 +112,7 @@
     "StixObjectOrStixRelationship",
     "StixSightingRelationship",
     "ThreatActor",
+    "ThreatActorGroup",
     "Tool",
     "Vulnerability",
     "get_config_variable",

pycti/api/opencti_api_client.py

@@ -59,6 +59,7 @@
 from pycti.entities.opencti_stix_sighting_relationship import StixSightingRelationship
 from pycti.entities.opencti_task import Task
 from pycti.entities.opencti_threat_actor import ThreatActor
+from pycti.entities.opencti_threat_actor_group import ThreatActorGroup
 from pycti.entities.opencti_tool import Tool
 from pycti.entities.opencti_vocabulary import Vocabulary
 from pycti.entities.opencti_vulnerability import Vulnerability
@@ -181,6 +182,7 @@ def __init__(
         self.event = Event(self)
         self.location = Location(self)
         self.threat_actor = ThreatActor(self)
+        self.threat_actor_group = ThreatActorGroup(self)
         self.intrusion_set = IntrusionSet(self)
         self.infrastructure = Infrastructure(self)
         self.campaign = Campaign(self)
@@ -407,7 +409,8 @@ def health_check(self):
             test = self.threat_actor.list(first=1)
             if test is not None:
                 return True
-        except:
+        except Exception as err:  # pylint: disable=broad-except
+            LOGGER.error("%s", err)
             return False
         return False
 

pycti/entities/opencti_stix_core_object.py

@@ -349,7 +349,9 @@ def __init__(self, opencti, file):
                 resource_level
                 primary_motivation
                 secondary_motivations
-                personal_motivations
+                ... on ThreatActorGroup {
+                    personal_motivations
+                }
             }
             ... on Tool {
                 name

pycti/entities/opencti_stix_domain_object.py

@@ -437,7 +437,9 @@ def __init__(self, opencti, file):
                 resource_level
                 primary_motivation
                 secondary_motivations
-                personal_motivations
+                ... on ThreatActorGroup {
+                    personal_motivations
+                }
             }
             ... on Tool {
                 name

pycti/entities/opencti_stix_object_or_stix_relationship.py

@@ -288,7 +288,9 @@ def __init__(self, opencti):
                 resource_level
                 primary_motivation
                 secondary_motivations
-                personal_motivations
+                ... on ThreatActorGroup {
+                    personal_motivations
+                }
             }
             ... on Tool {
                 name

pycti/entities/opencti_threat_actor.py

@@ -7,6 +7,7 @@
 from stix2.canonicalization.Canonicalize import canonicalize
 
 from pycti.entities import LOGGER
+from pycti.entities.opencti_threat_actor_group import ThreatActorGroup
 
 
 class ThreatActor:
@@ -19,6 +20,7 @@ def __init__(self, opencti):
         """Create an instance of ThreatActor"""
 
         self.opencti = opencti
+        self.threat_actor_group = ThreatActorGroup(opencti)
         self.properties = """
             id
             standard_id
@@ -130,7 +132,6 @@ def __init__(self, opencti):
             resource_level
             primary_motivation
             secondary_motivations
-            personal_motivations
             importFiles {
                 edges {
                     node {
@@ -299,87 +300,9 @@ def create(self, **kwargs):
         :param str resource_level: (optional) describe the actors resource_level in text
         :param str primary_motivation: (optional) describe the actors primary_motivation in text
         :param list secondary_motivations: (optional) describe the actors secondary_motivations in list of string
-        :param list personal_motivations: (optional) describe the actors personal_motivations in list of strings
         :param bool update: (optional) choose to updated an existing Threat-Actor entity, default `False`
         """
-
-        stix_id = kwargs.get("stix_id", None)
-        created_by = kwargs.get("createdBy", None)
-        object_marking = kwargs.get("objectMarking", None)
-        object_label = kwargs.get("objectLabel", None)
-        external_references = kwargs.get("externalReferences", None)
-        revoked = kwargs.get("revoked", None)
-        confidence = kwargs.get("confidence", None)
-        lang = kwargs.get("lang", None)
-        created = kwargs.get("created", None)
-        modified = kwargs.get("modified", None)
-        name = kwargs.get("name", None)
-        description = kwargs.get("description", None)
-        aliases = kwargs.get("aliases", None)
-        threat_actor_types = kwargs.get("threat_actor_types", None)
-        first_seen = kwargs.get("first_seen", None)
-        last_seen = kwargs.get("last_seen", None)
-        goals = kwargs.get("goals", None)
-        sophistication = kwargs.get("sophistication", None)
-        resource_level = kwargs.get("resource_level", None)
-        primary_motivation = kwargs.get("primary_motivation", None)
-        secondary_motivations = kwargs.get("secondary_motivations", None)
-        personal_motivations = kwargs.get("personal_motivations", None)
-        x_opencti_stix_ids = kwargs.get("x_opencti_stix_ids", None)
-        granted_refs = kwargs.get("objectOrganization", None)
-        update = kwargs.get("update", False)
-
-        if name is not None:
-            LOGGER.info("Creating Threat-Actor {%s}.", name)
-            query = """
-                mutation ThreatActorAdd($input: ThreatActorAddInput!) {
-                    threatActorAdd(input: $input) {
-                        id
-                        standard_id
-                        entity_type
-                        parent_types
-                    }
-                }
-            """
-            result = self.opencti.query(
-                query,
-                {
-                    "input": {
-                        "stix_id": stix_id,
-                        "createdBy": created_by,
-                        "objectMarking": object_marking,
-                        "objectLabel": object_label,
-                        "objectOrganization": granted_refs,
-                        "externalReferences": external_references,
-                        "revoked": revoked,
-                        "confidence": confidence,
-                        "lang": lang,
-                        "created": created,
-                        "modified": modified,
-                        "name": name,
-                        "description": description,
-                        "aliases": aliases,
-                        "threat_actor_types": threat_actor_types,
-                        "first_seen": first_seen,
-                        "last_seen": last_seen,
-                        "goals": goals,
-                        "sophistication": sophistication,
-                        "resource_level": resource_level,
-                        "primary_motivation": primary_motivation,
-                        "secondary_motivations": secondary_motivations,
-                        "personal_motivations": personal_motivations,
-                        "x_opencti_stix_ids": x_opencti_stix_ids,
-                        "update": update,
-                    }
-                },
-            )
-            return self.opencti.process_multiple_fields(
-                result["data"]["threatActorAdd"]
-            )
-        else:
-            LOGGER.error(
-                "[opencti_threat_actor] Missing parameters: name and description"
-            )
+        return self.threat_actor_group.create(**kwargs)
 
     """
         Import an Threat-Actor object from a STIX2 object
@@ -389,80 +312,4 @@ def create(self, **kwargs):
     """
 
     def import_from_stix2(self, **kwargs):
-        stix_object = kwargs.get("stixObject", None)
-        extras = kwargs.get("extras", {})
-        update = kwargs.get("update", False)
-        if stix_object is not None:
-            # Search in extensions
-            if "x_opencti_stix_ids" not in stix_object:
-                stix_object[
-                    "x_opencti_stix_ids"
-                ] = self.opencti.get_attribute_in_extension("stix_ids", stix_object)
-            if "granted_refs" not in stix_object:
-                stix_object["granted_refs"] = self.opencti.get_attribute_in_extension(
-                    "granted_refs", stix_object
-                )
-
-            return self.create(
-                stix_id=stix_object["id"],
-                createdBy=extras["created_by_id"]
-                if "created_by_id" in extras
-                else None,
-                objectMarking=extras["object_marking_ids"]
-                if "object_marking_ids" in extras
-                else None,
-                objectLabel=extras["object_label_ids"]
-                if "object_label_ids" in extras
-                else None,
-                externalReferences=extras["external_references_ids"]
-                if "external_references_ids" in extras
-                else None,
-                revoked=stix_object["revoked"] if "revoked" in stix_object else None,
-                confidence=stix_object["confidence"]
-                if "confidence" in stix_object
-                else None,
-                lang=stix_object["lang"] if "lang" in stix_object else None,
-                created=stix_object["created"] if "created" in stix_object else None,
-                modified=stix_object["modified"] if "modified" in stix_object else None,
-                name=stix_object["name"],
-                description=self.opencti.stix2.convert_markdown(
-                    stix_object["description"]
-                )
-                if "description" in stix_object
-                else None,
-                aliases=self.opencti.stix2.pick_aliases(stix_object),
-                threat_actor_types=stix_object["threat_actor_types"]
-                if "threat_actor_types" in stix_object
-                else None,
-                first_seen=stix_object["first_seen"]
-                if "first_seen" in stix_object
-                else None,
-                last_seen=stix_object["last_seen"]
-                if "last_seen" in stix_object
-                else None,
-                goals=stix_object["goals"] if "goals" in stix_object else None,
-                sophistication=stix_object["sophistication"]
-                if "sophistication" in stix_object
-                else None,
-                resource_level=stix_object["resource_level"]
-                if "resource_level" in stix_object
-                else None,
-                primary_motivation=stix_object["primary_motivation"]
-                if "primary_motivation" in stix_object
-                else None,
-                secondary_motivations=stix_object["secondary_motivations"]
-                if "secondary_motivations" in stix_object
-                else None,
-                personal_motivations=stix_object["personal_motivations"]
-                if "personal_motivations" in stix_object
-                else None,
-                x_opencti_stix_ids=stix_object["x_opencti_stix_ids"]
-                if "x_opencti_stix_ids" in stix_object
-                else None,
-                objectOrganization=stix_object["granted_refs"]
-                if "granted_refs" in stix_object
-                else None,
-                update=update,
-            )
-        else:
-            LOGGER.error("[opencti_threat_actor] Missing parameters: stixObject")
+        return self.threat_actor_group.import_from_stix2(**kwargs)