[client] Add sightings

Author: Samuel Hassine

examples/import_stix2_file.py

@@ -3,8 +3,8 @@
 from pycti import OpenCTIApiClient
 
 # Variables
-api_url = "https://demo.opencti.io"
-api_token = "2b4f29e3-5ea8-4890-8cf5-a76f61f1e2b2"
+api_url = "http://localhost:4000"
+api_token = "0b23f787-d013-41a8-8078-97bee84cc99d"
 
 # OpenCTI initialization
 opencti_api_client = OpenCTIApiClient(api_url, api_token)

pycti/__init__.py

@@ -18,6 +18,7 @@
 from .entities.opencti_stix_domain_entity import StixDomainEntity
 from .entities.opencti_stix_observable import StixObservable
 from .entities.opencti_stix_relation import StixRelation
+from .entities.opencti_stix_sighting import StixSighting
 from .entities.opencti_stix_observable_relation import StixObservableRelation
 from .entities.opencti_identity import Identity
 from .entities.opencti_threat_actor import ThreatActor
@@ -54,6 +55,7 @@
     "StixDomainEntity",
     "StixObservable",
     "StixRelation",
+    "StixSighting",
     "StixObservableRelation",
     "Identity",
     "ThreatActor",

pycti/api/opencti_api_client.py

@@ -20,6 +20,7 @@
 from pycti.entities.opencti_stix_domain_entity import StixDomainEntity
 from pycti.entities.opencti_stix_observable import StixObservable
 from pycti.entities.opencti_stix_relation import StixRelation
+from pycti.entities.opencti_stix_sighting import StixSighting
 from pycti.entities.opencti_stix_observable_relation import StixObservableRelation
 from pycti.entities.opencti_identity import Identity
 from pycti.entities.opencti_threat_actor import ThreatActor
@@ -89,6 +90,7 @@ def __init__(self, url, token, log_level="info", ssl_verify=False):
         self.stix_domain_entity = StixDomainEntity(self, File)
         self.stix_observable = StixObservable(self)
         self.stix_relation = StixRelation(self)
+        self.stix_sighting = StixSighting(self)
         self.stix_observable_relation = StixObservableRelation(self)
         self.identity = Identity(self)
         self.threat_actor = ThreatActor(self)

pycti/entities/opencti_attack_pattern.py

@@ -17,6 +17,7 @@ def __init__(self, opencti):
             name
             alias
             description
+            confidence
             graph_data
             platform
             required_permission
@@ -251,6 +252,7 @@ def create_raw(self, **kwargs):
         platform = kwargs.get("platform", None)
         required_permission = kwargs.get("required_permission", None)
         external_id = kwargs.get("external_id", None)
+        confidence = kwargs.get("confidence", 50)
         id = kwargs.get("id", None)
         stix_id_key = kwargs.get("stix_id_key", None)
         created = kwargs.get("created", None)
@@ -283,6 +285,7 @@ def create_raw(self, **kwargs):
                         "platform": platform,
                         "required_permission": required_permission,
                         "external_id": external_id,
+                        "confidence": confidence,
                         "internal_id_key": id,
                         "stix_id_key": stix_id_key,
                         "created": created,
@@ -317,6 +320,7 @@ def create(self, **kwargs):
         platform = kwargs.get("platform", None)
         required_permission = kwargs.get("required_permission", None)
         external_id = kwargs.get("external_id", None)
+        confidence = kwargs.get("confidence", 50)
         id = kwargs.get("id", None)
         stix_id_key = kwargs.get("stix_id_key", None)
         created = kwargs.get("created", None)
@@ -332,6 +336,7 @@ def create(self, **kwargs):
             name
             description 
             alias
+            confidence
             createdByRef {
                 node {
                     id
@@ -362,7 +367,10 @@ def create(self, **kwargs):
                     )
                     object_result["name"] = name
                 # description
-                if object_result["description"] != description:
+                if (
+                    description is not None
+                    and object_result["description"] != description
+                ):
                     self.opencti.stix_domain_entity.update_field(
                         id=object_result["id"], key="description", value=description
                     )
@@ -405,6 +413,12 @@ def create(self, **kwargs):
                         id=object_result["id"], key="external_id", value=external_id
                     )
                     object_result["external_id"] = external_id
+                # confidence
+                if confidence is not None and object_result["confidence"] != confidence:
+                    self.opencti.stix_domain_entity.update_field(
+                        id=object_result["id"], key="confidence", value=confidence
+                    )
+                    object_result["confidence"] = confidence
             return object_result
         else:
             return self.create_raw(
@@ -414,6 +428,7 @@ def create(self, **kwargs):
                 platform=platform,
                 required_permission=required_permission,
                 external_id=external_id,
+                confidence=confidence,
                 id=id,
                 stix_id_key=stix_id_key,
                 created=created,
@@ -465,6 +480,9 @@ def import_from_stix2(self, **kwargs):
                 if "x_mitre_permissions_required" in stix_object
                 else None,
                 external_id=external_id,
+                confidence=stix_object["confidence"]
+                if "confidence" in stix_object
+                else None,
                 id=stix_object[CustomProperties.ID]
                 if CustomProperties.ID in stix_object
                 else None,

pycti/entities/opencti_campaign.py

@@ -17,6 +17,7 @@ def __init__(self, opencti):
             name
             alias
             description
+            confidence
             graph_data
             objective
             first_seen
@@ -287,6 +288,7 @@ def create(self, **kwargs):
             name
             description 
             alias
+            confidence
             createdByRef {
                 node {
                     id
@@ -313,7 +315,10 @@ def create(self, **kwargs):
                     )
                     object_result["name"] = name
                 # description
-                if object_result["description"] != description:
+                if (
+                    description is not None
+                    and object_result["description"] != description
+                ):
                     self.opencti.stix_domain_entity.update_field(
                         id=object_result["id"], key="description", value=description
                     )
@@ -399,11 +404,11 @@ def to_stix2(self, **kwargs):
             if self.opencti.not_empty(entity["objective"]):
                 campaign["objective"] = entity["objective"]
             if self.opencti.not_empty(entity["first_seen"]):
-                campaign[CustomProperties.FIRST_SEEN] = self.opencti.stix2.format_date(
+                campaign["first_seen"] = self.opencti.stix2.format_date(
                     entity["first_seen"]
                 )
             if self.opencti.not_empty(entity["last_seen"]):
-                campaign[CustomProperties.LAST_SEEN] = self.opencti.stix2.format_date(
+                campaign["last_seen"] = self.opencti.stix2.format_date(
                     entity["last_seen"]
                 )
             campaign["created"] = self.opencti.stix2.format_date(entity["created"])

pycti/entities/opencti_course_of_action.py

@@ -17,6 +17,7 @@ def __init__(self, opencti):
             name
             alias
             description
+            confidence
             graph_data
             created
             modified            
@@ -302,7 +303,10 @@ def create(self, **kwargs):
                     )
                     object_result["name"] = name
                 # description
-                if object_result["description"] != description:
+                if (
+                    description is not None
+                    and object_result["description"] != description
+                ):
                     self.opencti.stix_domain_entity.update_field(
                         id=object_result["id"], key="description", value=description
                     )

pycti/entities/opencti_identity.py

@@ -302,7 +302,10 @@ def create(self, **kwargs):
                     )
                     object_result["name"] = name
                 # description
-                if object_result["description"] != description:
+                if (
+                    description is not None
+                    and object_result["description"] != description
+                ):
                     self.opencti.stix_domain_entity.update_field(
                         id=object_result["id"], key="description", value=description
                     )

pycti/entities/opencti_incident.py

@@ -321,7 +321,10 @@ def create(self, **kwargs):
                     )
                     object_result["name"] = name
                 # description
-                if object_result["description"] != description:
+                if (
+                    description is not None
+                    and object_result["description"] != description
+                ):
                     self.opencti.stix_domain_entity.update_field(
                         id=object_result["id"], key="description", value=description
                     )

pycti/entities/opencti_indicator.py

@@ -269,6 +269,8 @@ def create_raw(self, **kwargs):
         valid_from = kwargs.get("valid_from", None)
         valid_until = kwargs.get("valid_until", None)
         score = kwargs.get("score", None)
+        confidence = kwargs.get("confidence", 50)
+        detection = kwargs.get("detection", False)
         id = kwargs.get("id", None)
         stix_id_key = kwargs.get("stix_id_key", None)
         created = kwargs.get("created", None)
@@ -309,6 +311,8 @@ def create_raw(self, **kwargs):
                         "valid_from": valid_from,
                         "valid_until": valid_until,
                         "score": score,
+                        "detection": detection,
+                        "confidence": confidence,
                         "internal_id_key": id,
                         "stix_id_key": stix_id_key,
                         "created": created,
@@ -343,6 +347,8 @@ def create(self, **kwargs):
         valid_from = kwargs.get("valid_from", None)
         valid_until = kwargs.get("valid_until", None)
         score = kwargs.get("score", None)
+        confidence = kwargs.get("confidence", 50)
+        detection = kwargs.get("detection", False)
         id = kwargs.get("id", None)
         stix_id_key = kwargs.get("stix_id_key", None)
         created = kwargs.get("created", None)
@@ -358,11 +364,13 @@ def create(self, **kwargs):
             name
             description
             score
+            confidence
+            detection
             createdByRef {
                 node {
                     id
                 }
-            }                    
+            }
             ... on Indicator {
                 observableRefs {
                     edges {
@@ -418,6 +426,18 @@ def create(self, **kwargs):
                         id=object_result["id"], key="score", value=score
                     )
                     object_result["score"] = score
+                # confidence
+                if confidence is not None and object_result["confidence"] != confidence:
+                    self.opencti.stix_domain_entity.update_field(
+                        id=object_result["id"], key="confidence", value=confidence
+                    )
+                    object_result["confidence"] = confidence
+                # detection
+                if detection is not None and object_result["detection"] != detection:
+                    self.opencti.stix_domain_entity.update_field(
+                        id=object_result["id"], key="detection", value=detection
+                    )
+                    object_result["detection"] = detection
             return object_result
         else:
             return self.create_raw(
@@ -429,6 +449,8 @@ def create(self, **kwargs):
                 valid_from=valid_from,
                 valid_until=valid_until,
                 score=score,
+                detection=detection,
+                confidence=confidence,
                 id=id,
                 stix_id_key=stix_id_key,
                 created=created,
@@ -500,6 +522,75 @@ def add_stix_observable(self, **kwargs):
             )
             return False
 
+    """
+        Import an Indicator object from a STIX2 object
+
+        :param stixObject: the Stix-Object Indicator
+        :return Indicator object
+    """
+
+    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:
+            pattern_type = "stix"
+            if CustomProperties.PATTERN_TYPE in stix_object:
+                pattern_type = stix_object[CustomProperties.PATTERN_TYPE]
+            elif "pattern_type" in stix_object:
+                pattern_type = stix_object["pattern_type"]
+            return self.create(
+                name=stix_object["name"] if "name" in stix_object else "",
+                description=self.opencti.stix2.convert_markdown(
+                    stix_object["description"]
+                )
+                if "description" in stix_object
+                else "",
+                indicator_pattern=stix_object[CustomProperties.INDICATOR_PATTERN]
+                if CustomProperties.INDICATOR_PATTERN in stix_object
+                else stix_object["pattern"],
+                main_observable_type=stix_object[CustomProperties.OBSERVABLE_TYPE]
+                if CustomProperties.OBSERVABLE_TYPE in stix_object
+                else "Unknown",
+                pattern_type=pattern_type,
+                valid_from=stix_object["valid_from"]
+                if "valid_from" in stix_object
+                else None,
+                valid_until=stix_object["valid_until"]
+                if "valid_until" in stix_object
+                else None,
+                score=stix_object[CustomProperties.SCORE]
+                if CustomProperties.SCORE in stix_object
+                else None,
+                confidence=stix_object["confidence"]
+                if "confidence" in stix_object
+                else 50,
+                detection=stix_object[CustomProperties.DETECTION]
+                if CustomProperties.DETECTION in stix_object
+                else None,
+                id=stix_object[CustomProperties.ID]
+                if CustomProperties.ID in stix_object
+                else None,
+                stix_id_key=stix_object["id"] if "id" 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,
+                createdByRef=extras["created_by_ref_id"]
+                if "created_by_ref_id" in extras
+                else None,
+                markingDefinitions=extras["marking_definitions_ids"]
+                if "marking_definitions_ids" in extras
+                else None,
+                tags=extras["tags_ids"] if "tags_ids" in extras else [],
+                killChainPhases=extras["kill_chain_phases_ids"]
+                if "kill_chain_phases_ids" in extras
+                else [],
+                update=update,
+            )
+        else:
+            self.opencti.log(
+                "error", "[opencti_attack_pattern] Missing parameters: stixObject"
+            )
+
     """
         Export an Indicator object in STIX2
     
@@ -536,9 +627,9 @@ def to_stix2(self, **kwargs):
                 entity["valid_until"]
             )
             if self.opencti.not_empty(entity["pattern_type"]):
-                indicator[CustomProperties.PATTERN_TYPE] = entity["pattern_type"]
+                indicator["pattern_type"] = entity["pattern_type"]
             else:
-                indicator[CustomProperties.PATTERN_TYPE] = "stix"
+                indicator["pattern_type"] = "stix"
             indicator["created"] = self.opencti.stix2.format_date(entity["created"])
             indicator["modified"] = self.opencti.stix2.format_date(entity["modified"])
             if self.opencti.not_empty(entity["alias"]):