[client] Upgrade the Python client with new indicator type (OpenCTI-Platform#40)

Author: Samuel Hassine

LICENSE

@@ -3,6 +3,7 @@
 [![Website](https://img.shields.io/badge/website-opencti.io-blue.svg)](https://www.opencti.io)
 [![CircleCI](https://circleci.com/gh/OpenCTI-Platform/client-python.svg?style=shield)](https://circleci.com/gh/OpenCTI-Platform/client-python/tree/master)
 [![GitHub release](https://img.shields.io/github/release/OpenCTI-Platform/client-python.svg)](https://github.com/OpenCTI-Platform/client-python/releases/latest)
+[![Number of PyPI downloads](https://img.shields.io/pypi/dm/pycti.svg)](https://pypi.python.org/pypi/pycti/)
 [![Slack Status](https://slack.luatix.org/badge.svg)](https://slack.luatix.org)
 
 The official OpenCTI Python client helps developers to use the OpenCTI API by providing easy to use methods and utils. This client is also used by some OpenCTI components.

README.md

@@ -29,8 +29,6 @@
     toId=sector['id'],
     relationship_type='gathering',
     description='BNP Paribas is part of the sector Banking institutions.',
-    first_seen=datetime.datetime.today().strftime('%Y-%m-%dT%H:%M:%SZ'),
-    last_seen=datetime.datetime.today().strftime('%Y-%m-%dT%H:%M:%SZ'),
     ignore_dates=True
 )
 

examples/add_organization_to_sector.py

@@ -6,7 +6,7 @@
 from pycti import OpenCTIApiClient
 
 # Variables
-api_url = 'http://localhost:4000'
+api_url = 'https://demo.opencti.io'
 api_token = 'c2d944bb-aea6-4bd6-b3d7-6c10451e2256'
 
 # OpenCTI initialization
@@ -17,6 +17,7 @@
 
 # Prepare all the elements of the report
 object_refs = []
+observable_refs = []
 
 # Create the incident
 incident = opencti_api_client.incident.create(
@@ -26,7 +27,6 @@
 )
 print(incident)
 object_refs.append(incident['id'])
-
 # Create the associated report
 report = opencti_api_client.report.create(
     name="Report about my new incident",
@@ -58,29 +58,36 @@
         kill_chain_phase_id=kill_chain_phase_id
     )
 
+# Create the observable and indicator and indicates to the relation
 # Create the observable
 observable_ttp1 = opencti_api_client.stix_observable.create(
     type='Email-Address',
     observable_value='phishing@mail.com'
 )
+print(observable_ttp1)
+# Get the indicator
+indicator_ttp1 = observable_ttp1['indicators'][0]
+print(indicator_ttp1)
 # Indicates the relation Incident => uses => TTP
-observable_ttp1_relation = opencti_api_client.stix_relation.create(
-    fromType='Stix-Observable',
-    fromId=observable_ttp1['id'],
+indicator_ttp1_relation = opencti_api_client.stix_relation.create(
+    fromType='Indicator',
+    fromId=indicator_ttp1['id'],
     toType='stix_relation',
     toId=ttp1_relation['id'],
     relationship_type='indicates',
     description='This email address is the sender of the spearphishing.',
     first_seen=date,
     last_seen=date
 )
-# Elements for the report
+
+# Prepare elements for the report
 object_refs.extend([
     ttp1['id'],
     ttp1_relation['id'],
-    observable_ttp1['id'],
-    observable_ttp1_relation['id']
+    indicator_ttp1['id'],
+    indicator_ttp1_relation['id']
 ])
+observable_refs.append(observable_ttp1['id'])
 
 # Registry Run Keys / Startup Folder
 ttp2 = opencti_api_client.attack_pattern.read(filters=[{'key': 'external_id', 'values': ['T1060']}])
@@ -102,15 +109,21 @@
         id=ttp2_relation['id'],
         kill_chain_phase_id=kill_chain_phase_id
     )
-# Add observables to the relation
+
+# Create the observable and indicator and indicates to the relation
+# Create the observable
 observable_ttp2 = opencti_api_client.stix_observable.create(
     type='Registry-Key',
     observable_value='Disk security'
 )
+print(observable_ttp2)
+# Get the indicator
+indicator_ttp2 = observable_ttp2['indicators'][0]
+print(indicator_ttp2)
 # Indicates the relation Incident => uses => TTP
-observable_ttp2_relation = opencti_api_client.stix_relation.create(
-    fromType='Stix-Observable',
-    fromId=observable_ttp2['id'],
+indicator_ttp2_relation = opencti_api_client.stix_relation.create(
+    fromType='Indicator',
+    fromId=indicator_ttp2['id'],
     toType='stix_relation',
     toId=ttp2_relation['id'],
     relationship_type='indicates',
@@ -122,9 +135,10 @@
 object_refs.extend([
     ttp2['id'],
     ttp2_relation['id'],
-    observable_ttp2['id'],
-    observable_ttp2_relation['id']
+    indicator_ttp2['id'],
+    indicator_ttp2_relation['id']
 ])
+observable_refs.append(observable_ttp2['id'])
 
 # Data Encrypted
 ttp3 = opencti_api_client.attack_pattern.read(filters=[{'key': 'external_id', 'values': ['T1022']}])
@@ -151,3 +165,5 @@
 # Add all element to the report
 for object_ref in object_refs:
     opencti_api_client.report.add_stix_entity(id=report['id'], report=report, entity_id=object_ref)
+for observable_ref in observable_refs:
+    opencti_api_client.report.add_stix_observable(id=report['id'], report=report, entity_id=observable_ref)

examples/create_incident_with_ttps_and_observables.py examples/create_incident_with_ttps_and_indicators.py

@@ -4,7 +4,7 @@
 
 from pycti.connector.opencti_connector import ConnectorType
 from pycti.connector.opencti_connector import OpenCTIConnector
-from pycti.connector.opencti_connector_helper import OpenCTIConnectorHelper
+from pycti.connector.opencti_connector_helper import OpenCTIConnectorHelper, get_config_variable
 
 from pycti.entities.opencti_marking_definition import MarkingDefinition
 from pycti.entities.opencti_external_reference import ExternalReference
@@ -25,6 +25,7 @@
 from pycti.entities.opencti_attack_pattern import AttackPattern
 from pycti.entities.opencti_course_of_action import CourseOfAction
 from pycti.entities.opencti_report import Report
+from pycti.entities.opencti_indicator import Indicator
 
 from pycti.utils.opencti_stix2 import OpenCTIStix2
 from pycti.utils.constants import ObservableTypes

pycti/__init__.py

@@ -35,6 +35,7 @@
 from pycti.entities.opencti_attack_pattern import AttackPattern
 from pycti.entities.opencti_course_of_action import CourseOfAction
 from pycti.entities.opencti_report import Report
+from pycti.entities.opencti_indicator import Indicator
 
 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
 
@@ -96,6 +97,7 @@ def __init__(self, url, token, log_level='info', ssl_verify=False):
         self.attack_pattern = AttackPattern(self)
         self.course_of_action = CourseOfAction(self)
         self.report = Report(self)
+        self.indicator = Indicator(self)
 
         # Check if openCTI is available
         if not self.health_check():
@@ -275,6 +277,9 @@ def process_multiple_fields(self, data):
         if 'stixRelations' in data:
             data['stixRelations'] = self.process_multiple(data['stixRelations'])
             data['stixRelationsIds'] = self.process_multiple_ids(data['stixRelations'])
+        if 'indicators' in data:
+            data['indicators'] = self.process_multiple(data['indicators'])
+            data['indicatorsIds'] = self.process_multiple_ids(data['indicators'])
         return data
 
     @deprecated(version='2.1.0', reason="Replaced by the StixDomainEntity class in pycti")
@@ -1754,11 +1759,9 @@ def resolve_role(self, relation_type, from_type, to_type):
 
         relation_type = relation_type.lower()
         from_type = from_type.lower()
-        from_type = 'observable' if (
-                (ObservableTypes.has_value(from_type) and (
-                        relation_type == 'indicates' or relation_type == 'localization' or relation_type == 'gathering')
-                 ) or from_type == 'stix-observable'
-        ) else from_type
+        from_type = 'observable' if ((ObservableTypes.has_value(from_type) and (
+                    relation_type == 'localization' or relation_type == 'gathering')) or from_type == 'stix-observable'
+                                     ) else from_type
         to_type = to_type.lower()
         mapping = {
             'uses': {
@@ -1903,11 +1906,10 @@ def resolve_role(self, relation_type, from_type, to_type):
                 },
             },
             'indicates': {
-                'observable': {
+                'indicator': {
                     'threat-actor': {'from_role': 'indicator', 'to_role': 'characterize'},
                     'intrusion-set': {'from_role': 'indicator', 'to_role': 'characterize'},
                     'campaign': {'from_role': 'indicator', 'to_role': 'characterize'},
-                    'incident': {'from_role': 'indicator', 'to_role': 'characterize'},
                     'malware': {'from_role': 'indicator', 'to_role': 'characterize'},
                     'tool': {'from_role': 'indicator', 'to_role': 'characterize'},
                     'stix_relation': {'from_role': 'indicator', 'to_role': 'characterize'},

pycti/api/opencti_api_client.py

@@ -14,6 +14,27 @@
 from pycti.connector.opencti_connector import OpenCTIConnector
 
 
+def get_config_variable(envvar, yaml_path, config={}, isNumber=False):
+    if os.getenv(envvar) is not None:
+        result = os.getenv(envvar)
+    elif yaml_path is not None:
+        if yaml_path[0] in config and yaml_path[1] in config[yaml_path[0]]:
+            result = config[yaml_path[0]][yaml_path[1]]
+        else:
+            return None
+    else:
+        return None
+
+    if result == 'yes' or result == 'true' or result == 'True':
+        return True
+    elif result == 'no' or result == 'false' or result == 'False':
+        return False
+    elif isNumber:
+        return int(result)
+    else:
+        return result
+
+
 class ListenQueue(threading.Thread):
     def __init__(self, helper, config, callback):
         threading.Thread.__init__(self)
@@ -104,18 +125,22 @@ class OpenCTIConnectorHelper:
 
     def __init__(self, config: dict):
         # Load API config
-        self.opencti_url = os.getenv('OPENCTI_URL') or config['opencti']['url']
-        self.opencti_token = os.getenv('OPENCTI_TOKEN') or config['opencti']['token']
+        self.opencti_url = get_config_variable('OPENCTI_URL', ['opencti', 'url'], config)
+        self.opencti_token = get_config_variable('OPENCTI_TOKEN', ['opencti', 'token'], config)
         # Load connector config
-        self.connect_id = os.getenv('CONNECTOR_ID') or config['connector']['id']
-        self.connect_type = os.getenv('CONNECTOR_TYPE') or config['connector']['type']
-        self.connect_name = os.getenv('CONNECTOR_NAME') or config['connector']['name']
-        self.connect_confidence_level = int(
-            os.getenv('CONNECTOR_CONFIDENCE_LEVEL') or config['connector']['confidence_level'] or 2)
-        self.connect_scope = os.getenv('CONNECTOR_SCOPE') or config['connector']['scope']
-        self.log_level = os.getenv('CONNECTOR_LOG_LEVEL') or config['connector']['log_level'] or 'info'
-
-        # Configure logger²
+        self.connect_id = get_config_variable('CONNECTOR_ID', ['connector', 'id'], config)
+        self.connect_type = get_config_variable('CONNECTOR_TYPE', ['connector', 'type'], config)
+        self.connect_name = get_config_variable('CONNECTOR_NAME', ['connector', 'name'], config)
+        self.connect_confidence_level = get_config_variable(
+            'CONNECTOR_CONFIDENCE_LEVEL',
+            ['connector', 'confidence_level'],
+            config,
+            True
+        )
+        self.connect_scope = get_config_variable('CONNECTOR_SCOPE', ['connector', 'scope'], config)
+        self.log_level = get_config_variable('CONNECTOR_LOG_LEVEL', ['connector', 'log_level'], config)
+
+        # Configure logger
         numeric_level = getattr(logging, self.log_level.upper(), None)
         if not isinstance(numeric_level, int):
             raise ValueError('Invalid log level: ' + self.log_level)

pycti/connector/opencti_connector_helper.py

@@ -255,6 +255,8 @@ def create(self, **kwargs):
             stix_id_key=stix_id_key,
             name=name
         )
+        if object_result is None and external_id is not None:
+            object_result = self.read(filters=[{'key': 'external_id', 'values': [external_id]}])
         if object_result is not None:
             if update:
                 # name
@@ -326,6 +328,8 @@ def import_from_stix2(self, **kwargs):
         if stix_object is not None:
             # Extract external ID
             external_id = None
+            if CustomProperties.EXTERNAL_ID in stix_object:
+                external_id = stix_object[CustomProperties.EXTERNAL_ID]
             if 'external_references' in stix_object:
                 for external_reference in stix_object['external_references']:
                     if external_reference['source_name'] == 'mitre-attack' or external_reference[
@@ -368,6 +372,7 @@ def to_stix2(self, **kwargs):
             attack_pattern = dict()
             attack_pattern['id'] = entity['stix_id_key']
             attack_pattern['type'] = 'attack-pattern'
+            if self.opencti.not_empty(entity['external_id']): attack_pattern[CustomProperties.EXTERNAL_ID] = entity['external_id']
             attack_pattern['name'] = entity['name']
             if self.opencti.not_empty(entity['stix_label']):
                 attack_pattern['labels'] = entity['stix_label']