[client/examples] End of work about indicators/observables

Author: Samuel Hassine

examples/create_campaign_attributed-to_intrusion_set.py

@@ -1,7 +1,5 @@
 # coding: utf-8
 
-import datetime
-
 from dateutil.parser import parse
 from pycti import OpenCTIApiClient
 

examples/create_indicator_of_campaign.py

@@ -0,0 +1,70 @@
+# coding: utf-8
+
+import datetime
+
+from dateutil.parser import parse
+from pycti import OpenCTIApiClient
+
+# Variables
+api_url = 'https://demo.opencti.io'
+api_token = 'c2d944bb-aea6-4bd6-b3d7-6c10451e2256'
+
+# OpenCTI initialization
+opencti_api_client = OpenCTIApiClient(api_url, api_token)
+
+# Define the date
+date = parse('2019-12-01').strftime('%Y-%m-%dT%H:%M:%SZ')
+
+# Create the Campaign
+campaign = opencti_api_client.campaign.create(
+    name='My new Campaign',
+    description='Large SpearPhishing and intrusions followed by ransomware',
+    objective='Financial gain',
+    first_seen=date,
+    last_seen=date,
+    update=True
+)
+print(campaign)
+
+# Create the indicator
+indicator = opencti_api_client.indicator.create(
+    name='C2 server of the new campaign',
+    description='This is the C2 server of the campaign',
+    pattern_type='stix',
+    indicator_pattern="[domain-name:value = 'www.5z8.info' AND domain-name:resolves_to_refs[*].value = '198.51.100.1/32']",
+    main_observable_type='IPv4-Addr',
+    valid_from=date
+)
+print(indicator)
+
+# Create the observables (optional)
+observable_1 = opencti_api_client.stix_observable.create(
+    type='Domain',
+    observable_value='www.5z8.info'
+)
+observable_2 = opencti_api_client.stix_observable.create(
+    type='IPv4-Addr',
+    observable_value='198.51.100.1'
+)
+# Create the relation between observables and the indicator
+opencti_api_client.indicator.add_stix_observable(
+    id=indicator['id'],
+    stix_observable_id=observable_1['id']
+)
+opencti_api_client.indicator.add_stix_observable(
+    id=indicator['id'],
+    stix_observable_id=observable_2['id']
+)
+
+# Create the relation
+relation = opencti_api_client.stix_relation.create(
+    fromType='Indicator',
+    fromId=indicator['id'],
+    toType='Campaign',
+    toId=campaign['id'],
+    relationship_type='indicates',
+    first_seen=date,
+    last_seen=date,
+    description='This is the C2 server of the campaign.'
+)
+print(relation)

pycti/entities/opencti_indicator.py

@@ -206,6 +206,7 @@ def create_raw(self, **kwargs):
         name = kwargs.get('name', None)
         description = kwargs.get('description', None)
         indicator_pattern = kwargs.get('indicator_pattern', None)
+        main_observable_type = kwargs.get('main_observable_type', None)
         pattern_type = kwargs.get('pattern_type', None)
         valid_from = kwargs.get('valid_from', None)
         valid_until = kwargs.get('valid_until', None)
@@ -216,7 +217,7 @@ def create_raw(self, **kwargs):
         modified = kwargs.get('modified', None)
         created_by_ref = kwargs.get('createdByRef', None)
 
-        if name is not None and indicator_pattern is not None:
+        if name is not None and indicator_pattern is not None and main_observable_type is not None:
             self.opencti.log('info', 'Creating Indicator {' + name + '}.')
             query = """
                 mutation IndicatorAdd($input: IndicatorAddInput) {
@@ -230,6 +231,7 @@ def create_raw(self, **kwargs):
                     'name': name,
                     'description': description,
                     'indicator_pattern': indicator_pattern,
+                    'main_observable_type': main_observable_type,
                     'pattern_type': pattern_type,
                     'valid_from': valid_from,
                     'valid_until': valid_until,
@@ -243,7 +245,10 @@ def create_raw(self, **kwargs):
             })
             return self.opencti.process_multiple_fields(result['data']['indicatorAdd'])
         else:
-            self.opencti.log('error', '[opencti_indicator] Missing parameters: name and indicator_pattern')
+            self.opencti.log(
+                'error',
+                '[opencti_indicator] Missing parameters: name and indicator_pattern and main_observable_type'
+            )
 
     """
         Create a Indicator object only if it not exists, update it on request
@@ -256,6 +261,7 @@ def create(self, **kwargs):
         name = kwargs.get('name', None)
         description = kwargs.get('description', None)
         indicator_pattern = kwargs.get('indicator_pattern', None)
+        main_observable_type = kwargs.get('main_observable_type', None)
         pattern_type = kwargs.get('pattern_type', None)
         valid_from = kwargs.get('valid_from', None)
         valid_until = kwargs.get('valid_until', None)
@@ -287,6 +293,7 @@ def create(self, **kwargs):
                 name=name,
                 description=description,
                 indicator_pattern=indicator_pattern,
+                main_observable_type=main_observable_type,
                 pattern_type=pattern_type,
                 valid_from=valid_from,
                 valid_until=valid_until,
@@ -297,6 +304,54 @@ def create(self, **kwargs):
                 createdByRef=created_by_ref
             )
 
+    """
+        Add a Stix-Observable object to Indicator object (observable_refs)
+
+        :param id: the id of the Indicator
+        :param entity_id: the id of the Stix-Observable
+        :return Boolean
+    """
+
+    def add_stix_observable(self, **kwargs):
+        id = kwargs.get('id', None)
+        indicator = kwargs.get('indicator', None)
+        stix_observable_id = kwargs.get('stix_observable_id', None)
+        if id is not None and stix_observable_id is not None:
+            if indicator is None:
+                indicator = self.read(id=id)
+            if indicator is None:
+                self.opencti.log('error', '[opencti_indicator] Cannot add Object Ref, indicator not found')
+                return False
+            if stix_observable_id in indicator['observableRefsIds']:
+                return True
+            else:
+                self.opencti.log(
+                    'info',
+                    'Adding Stix-Observable {' + stix_observable_id + '} to Indicator {' + id + '}'
+                )
+                query = """
+                   mutation IndicatorEdit($id: ID!, $input: RelationAddInput) {
+                       indicatorEdit(id: $id) {
+                            relationAdd(input: $input) {
+                                id
+                            }
+                       }
+                   }
+                """
+                self.opencti.query(query, {
+                    'id': id,
+                    'input': {
+                        'fromRole': 'observables_aggregation',
+                        'toId': stix_observable_id,
+                        'toRole': 'soo',
+                        'through': 'observable_refs'
+                    }
+                })
+                return True
+        else:
+            self.opencti.log('error', '[opencti_indicator] Missing parameters: id and stix_observable_id')
+            return False
+
     """
         Export an Indicator object in STIX2
     

pycti/entities/opencti_stix_relation.py

@@ -501,8 +501,8 @@ def to_stix2(self, **kwargs):
             if self.opencti.not_empty(entity['description']): stix_relation['description'] = entity['description']
             stix_relation['source_ref'] = entity['from']['stix_id_key']
             stix_relation['target_ref'] = entity['to']['stix_id_key']
-            stix_relation[CustomProperties.SOURCE_REF] = entity['from']['id']
-            stix_relation[CustomProperties.TARGET_REF] = entity['to']['id']
+            stix_relation[CustomProperties.SOURCE_REF] = entity['from']['stix_id_key']
+            stix_relation[CustomProperties.TARGET_REF] = entity['to']['stix_id_key']
             stix_relation['created'] = self.opencti.stix2.format_date(entity['created'])
             stix_relation['modified'] = self.opencti.stix2.format_date(entity['modified'])
             if self.opencti.not_empty(entity['first_seen']): stix_relation[

pycti/utils/opencti_stix2.py

@@ -755,14 +755,14 @@ def export_entity(self, entity_type, entity_id, mode='simple', max_marking_defin
             mode=mode,
             max_marking_definition_entity=max_marking_definition_entity
         )
-        for object in objects:
-            object['id'] = object['id'].replace('observable', 'indicator')
-            if 'source_ref' in object:
-                object['source_ref'] = object['source_ref'].replace('observable', 'indicator')
-            if 'target_ref' in object:
-                object['target_ref'] = object['target_ref'].replace('observable', 'indicator')
-            bundle['objects'].append(object)
-
+        if objects is not None:
+            for object in objects:
+                object['id'] = object['id'].replace('observable', 'indicator')
+                if 'source_ref' in object:
+                    object['source_ref'] = object['source_ref'].replace('observable', 'indicator')
+                if 'target_ref' in object:
+                    object['target_ref'] = object['target_ref'].replace('observable', 'indicator')
+                bundle['objects'].append(object)
         return bundle
 
     def export_bundle(self, types=[]):
@@ -1280,6 +1280,8 @@ def create_indicator(self, stix_object, update=False):
             name=stix_object['name'],
             description=self.convert_markdown(stix_object['description']) if 'description' in stix_object else '',
             indicator_pattern=stix_object['pattern'],
+            main_observable_type=stix_object[
+                CustomProperties.OBSERVABLE_TYPE] if CustomProperties.OBSERVABLE_TYPE in stix_object else 'Unknown',
             pattern_type=stix_object[
                 CustomProperties.PATTERN_TYPE] if CustomProperties.PATTERN_TYPE in stix_object else 'stix2',
             valid_from=stix_object['valid_from'] if 'valid_from' in stix_object else None,
@@ -1420,11 +1422,25 @@ def import_bundle(self, stix_bundle, update=False, types=None) -> List:
         start_time = time.time()
         for item in stix_bundle['objects']:
             if item['type'] == 'relationship':
-                self.import_relationship(item, update, types)
-                imported_elements.append({'id': item['id'], 'type': item['type']})
+                # Import only relationships between entities
+                if 'relationship' not in item[CustomProperties.SOURCE_REF] and \
+                        'relationship' not in item[CustomProperties.TARGET_REF]:
+                    self.import_relationship(item, update, types)
+                    imported_elements.append({'id': item['id'], 'type': item['type']})
         end_time = time.time()
         self.opencti.log('info', "Relationships imported in: %ssecs" % round(end_time - start_time))
 
+        # StixRelationObjects (with relationships)
+        start_time = time.time()
+        for item in stix_bundle['objects']:
+            if item['type'] == 'relationship':
+                if 'relationship' in item[CustomProperties.SOURCE_REF] or \
+                        'relationship' in item[CustomProperties.TARGET_REF]:
+                    self.import_relationship(item, update, types)
+                    imported_elements.append({'id': item['id'], 'type': item['type']})
+        end_time = time.time()
+        self.opencti.log('info', "Relationships to relationships imported in: %ssecs" % round(end_time - start_time))
+
         # StixCyberObservables
         start_time = time.time()
         for item in stix_bundle['objects']: