[examples] Fix examples and add examples to drone pipeline (OpenCTI-Platform#286)

Author: nor3th

.drone.yml

@@ -26,6 +26,15 @@ steps:
       - pip3 install -r test-requirements.txt --user
       - python3 -m pytest --no-header -vv --disable-warnings --cov=pycti --drone
 
+  # always run the examples last since they don't clean up
+  - name: example-tests
+    image: python:3.10
+    commands:
+      - pip3 install -r requirements.txt --user
+      - pip3 install .
+      - cd examples/
+      - /bin/bash run_all.sh
+
   - name: slack
     image: plugins/slack
     settings:

examples/add_external_reference_to_report.py

@@ -5,8 +5,8 @@
 from pycti import OpenCTIApiClient
 
 # Variables
-api_url = "https://demo.opencti.io"
-api_token = "YOUR_TOKEN"
+api_url = "http://opencti:4000"
+api_token = "bfa014e0-e02e-4aa6-a42b-603b19dcf159"
 
 # OpenCTI initialization
 opencti_api_client = OpenCTIApiClient(api_url, api_token)

examples/add_label_to_malware.py

@@ -3,8 +3,8 @@
 from pycti import OpenCTIApiClient
 
 # Variables
-api_url = "https://demo.opencti.io"
-api_token = "YOUR_TOKEN"
+api_url = "http://opencti:4000"
+api_token = "bfa014e0-e02e-4aa6-a42b-603b19dcf159"
 
 # OpenCTI initialization
 opencti_api_client = OpenCTIApiClient(api_url, api_token)

examples/add_label_to_observable.py

@@ -3,8 +3,8 @@
 from pycti import OpenCTIApiClient
 
 # Variables
-api_url = "https://demo.opencti.io"
-api_token = "YOUR_TOKEN"
+api_url = "http://opencti:4000"
+api_token = "bfa014e0-e02e-4aa6-a42b-603b19dcf159"
 
 # OpenCTI initialization
 opencti_api_client = OpenCTIApiClient(api_url, api_token)

examples/add_organization_to_sector.py

@@ -3,15 +3,15 @@
 from pycti import OpenCTIApiClient
 
 # Variables
-api_url = "https://demo.opencti.io"
-api_token = "YOUR_TOKEN"
+api_url = "http://opencti:4000"
+api_token = "bfa014e0-e02e-4aa6-a42b-603b19dcf159"
 
 # OpenCTI initialization
 opencti_api_client = OpenCTIApiClient(api_url, api_token)
 
 # Get the sector
-sector = opencti_api_client.identity.read(
-    filters=[{"key": "name", "values": ["Banking institutions"]}]
+sector = opencti_api_client.identity.create(
+    type="Sector", name="Banking institutions", description="Banks"
 )
 
 # Create the organization

examples/add_tool_usage_to_intrusion-set.py

@@ -3,8 +3,8 @@
 from pycti import OpenCTIApiClient
 
 # Variables
-api_url = "https://demo.opencti.io"
-api_token = "YOUR_TOKEN"
+api_url = "http://opencti:4000"
+api_token = "bfa014e0-e02e-4aa6-a42b-603b19dcf159"
 
 # OpenCTI initialization
 opencti_api_client = OpenCTIApiClient(api_url, api_token)

examples/ask_enrichment_of_observable.py

@@ -2,8 +2,8 @@
 from pycti import OpenCTIApiClient
 
 # Variables
-api_url = "https://demo.opencti.io"
-api_token = "YOUR_TOKEN"
+api_url = "http://opencti:4000"
+api_token = "bfa014e0-e02e-4aa6-a42b-603b19dcf159"
 # Define name of INTERNAL_ENRICHMENT Connector which can enrich IPv4 addresses
 connector_name = "AbuseIPDB"
 

examples/cmd_line_tag_latest_indicators_of_threat.py

@@ -6,104 +6,109 @@
 from pycti import OpenCTIApiClient
 
 # Variables
-api_url = "https://demo.opencti.io"
-api_token = "YOUR_TOKEN"
+api_url = "http://opencti:4000"
+api_token = "bfa014e0-e02e-4aa6-a42b-603b19dcf159"
 
 # OpenCTI initialization
 opencti_api_client = OpenCTIApiClient(api_url, api_token)
 
-# Parameters
-parser = argparse.ArgumentParser(description="Mandatory arguments.")
-parser.add_argument(
-    "--entity-type",
-    dest="entity_type",
-    default="Intrusion-Set",
-    required=True,
-    help="Type of the threat (Threat-Actor, Intrusion-Set, Campaign, X-OpenCTI,-Incident, Malware, Tool, Attack-Pattern)",
-)
-parser.add_argument(
-    "--name",
-    dest="name",
-    required=True,
-    help="Name of the threat",
-)
-parser.add_argument(
-    "--created-after",
-    dest="createdAfter",
-    help="Indicator created before (ISO date)",
-)
-parser.add_argument(
-    "--created-before",
-    dest="createdBefore",
-    help="Indicator created after (ISO date)",
-)
-parser.add_argument(
-    "--tags",
-    dest="tags",
-    required=True,
-    help="Tags to add or remove (separated by ,)",
-)
-parser.add_argument(
-    "--operation",
-    dest="operation",
-    required=True,
-    default="add",
-    help="Operation (add/remove)",
-)
-args = parser.parse_args()
 
-entity_type = args.entity_type
-name = args.name
-created_after = parse(args.createdAfter).strftime("%Y-%m-%dT%H:%M:%SZ")
-created_before = parse(args.createdBefore).strftime("%Y-%m-%dT%H:%M:%SZ")
-tags = args.tags.split(",")
-operation = args.operation
+def main():
+    # Parameters
+    parser = argparse.ArgumentParser(description="Mandatory arguments.")
+    parser.add_argument(
+        "--entity-type",
+        dest="entity_type",
+        default="Intrusion-Set",
+        required=True,
+        help="Type of the threat (Threat-Actor, Intrusion-Set, Campaign, X-OpenCTI,-Incident, Malware, Tool, Attack-Pattern)",
+    )
+    parser.add_argument(
+        "--name",
+        dest="name",
+        required=True,
+        help="Name of the threat",
+    )
+    parser.add_argument(
+        "--created-after",
+        dest="createdAfter",
+        help="Indicator created before (ISO date)",
+    )
+    parser.add_argument(
+        "--created-before",
+        dest="createdBefore",
+        help="Indicator created after (ISO date)",
+    )
+    parser.add_argument(
+        "--tags",
+        dest="tags",
+        required=True,
+        help="Tags to add or remove (separated by ,)",
+    )
+    parser.add_argument(
+        "--operation",
+        dest="operation",
+        required=True,
+        default="add",
+        help="Operation (add/remove)",
+    )
+    args = parser.parse_args()
 
-# Resolve the entity
-threat = opencti_api_client.stix_domain_object.read(
-    types=[entity_type], filters=[{"key": "name", "values": [name]}]
-)
+    entity_type = args.entity_type
+    name = args.name
+    created_after = parse(args.createdAfter).strftime("%Y-%m-%dT%H:%M:%SZ")
+    created_before = parse(args.createdBefore).strftime("%Y-%m-%dT%H:%M:%SZ")
+    tags = args.tags.split(",")
+    operation = args.operation
 
-if not threat:
-    raise ValueError("Cannot find the entity with the name " + name)
+    # Resolve the entity
+    threat = opencti_api_client.stix_domain_object.read(
+        types=[entity_type], filters=[{"key": "name", "values": [name]}]
+    )
 
-# Resolve all tags
-labels = []
-for tag in tags:
-    labels.append(opencti_api_client.label.create(value=tag))
+    if not threat:
+        raise ValueError("Cannot find the entity with the name " + name)
 
-# Get indicators
-custom_attributes = """
-    id
-    created_at
-"""
+    # Resolve all tags
+    labels = []
+    for tag in tags:
+        labels.append(opencti_api_client.label.create(value=tag))
 
+    # Get indicators
+    custom_attributes = """
+        id
+        created_at
+    """
 
-data = {"pagination": {"hasNextPage": True, "endCursor": None}}
-while data["pagination"]["hasNextPage"]:
-    after = data["pagination"]["endCursor"]
-    data = opencti_api_client.indicator.list(
-        first=50,
-        after=after,
-        customAttributes=custom_attributes,
-        filters=[
-            {"key": "indicates", "values": [threat["id"]]},
-            {"key": "created_at", "values": [created_after], "operator": "gt"},
-            {"key": "created_at", "values": [created_before], "operator": "lt"},
-        ],
-        orderBy="created_at",
-        orderMode="asc",
-        withPagination=True,
-    )
-    for indicator in data["entities"]:
-        print("[" + indicator["created_at"] + "] " + indicator["id"])
-        if operation == "add":
-            for label in labels:
-                opencti_api_client.stix_domain_object.add_label(
-                    id=indicator["id"], label_id=label["id"]
-                )
-        elif operation == "remove":
-            for label in labels:
-                opencti_api_client.stix_domain_object.remove_label(
-                    id=indicator["id"], label_id=label["id"]
-                )
+    data = {"pagination": {"hasNextPage": True, "endCursor": None}}
+    while data["pagination"]["hasNextPage"]:
+        after = data["pagination"]["endCursor"]
+        data = opencti_api_client.indicator.list(
+            first=50,
+            after=after,
+            customAttributes=custom_attributes,
+            filters=[
+                {"key": "indicates", "values": [threat["id"]]},
+                {"key": "created_at", "values": [created_after], "operator": "gt"},
+                {"key": "created_at", "values": [created_before], "operator": "lt"},
+            ],
+            orderBy="created_at",
+            orderMode="asc",
+            withPagination=True,
+        )
+        for indicator in data["entities"]:
+            print("[" + indicator["created_at"] + "] " + indicator["id"])
+            if operation == "add":
+                for label in labels:
+                    opencti_api_client.stix_domain_object.add_label(
+                        id=indicator["id"], label_id=label["id"]
+                    )
+            elif operation == "remove":
+                for label in labels:
+                    opencti_api_client.stix_domain_object.remove_label(
+                        id=indicator["id"], label_id=label["id"]
+                    )
+
+
+if __name__ == "__main__":
+    main()

examples/create_campaign_attributed-to_intrusion_set.py

@@ -5,8 +5,8 @@
 from pycti import OpenCTIApiClient
 
 # Variables
-api_url = "https://demo.opencti.io"
-api_token = "YOUR_TOKEN"
+api_url = "http://opencti:4000"
+api_token = "bfa014e0-e02e-4aa6-a42b-603b19dcf159"
 
 # OpenCTI initialization
 opencti_api_client = OpenCTIApiClient(api_url, api_token)

examples/create_file_with_hashes.py

@@ -3,8 +3,8 @@
 from pycti import OpenCTIApiClient
 
 # Variables
-api_url = "https://demo.opencti.io"
-api_token = "98481988-5aac-42e3-9be1-e1328ef86419"
+api_url = "http://opencti:4000"
+api_token = "bfa014e0-e02e-4aa6-a42b-603b19dcf159"
 
 # OpenCTI initialization
 opencti_api_client = OpenCTIApiClient(api_url, api_token)

examples/create_incident_with_ttps_and_indicators.py

@@ -5,8 +5,8 @@
 from pycti import OpenCTIApiClient
 
 # Variables
-api_url = "https://demo.opencti.io"
-api_token = "YOUR_TOKEN"
+api_url = "http://opencti:4000"
+api_token = "bfa014e0-e02e-4aa6-a42b-603b19dcf159"
 
 # OpenCTI initialization
 opencti_api_client = OpenCTIApiClient(api_url, api_token)
@@ -35,10 +35,17 @@
 
 # Associate the TTPs to the incident
 
-# Spearphishing Attachment
-ttp1 = opencti_api_client.attack_pattern.read(
-    filters=[{"key": "x_mitre_id", "values": ["T1193"]}]
+kcp_ia = opencti_api_client.kill_chain_phase.create(
+    phase_name="initial-access", kill_chain_name="mitre-attack"
 )
+
+ttp1 = opencti_api_client.attack_pattern.create(
+    name="Phishing: Spearphishing Attachment",
+    description="Adversaries may send spearphishing emails with a malicious attachment in an attempt to gain access to victim systems. Spearphishing attachment is a specific variant of spearphishing. Spearphishing attachment is different from other forms of spearphishing in that it employs the use of malware attached to an email. All forms of spearphishing are electronically delivered social engineering targeted at a specific individual, company, or industry. In this scenario, adversaries attach a file to the spearphishing email and usually rely upon User Execution to gain execution. Spearphishing may also involve social engineering techniques, such as posing as a trusted source.",
+    x_mitre_id="T1566.001",
+    killChainPhases=[kcp_ia["id"]],
+)
+ttp1 = opencti_api_client.attack_pattern.read(id=ttp1["id"])
 ttp1_relation = opencti_api_client.stix_core_relationship.create(
     fromId=incident["id"],
     toId=ttp1["id"],
@@ -53,7 +60,6 @@
         id=ttp1_relation["id"], 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_cyber_observable.create(
@@ -84,10 +90,21 @@
 )
 observable_refs.append(observable_ttp1["id"])
 
+kcp_p = opencti_api_client.kill_chain_phase.create(
+    phase_name="persistence", kill_chain_name="mitre-attack"
+)
+kcp_pe = opencti_api_client.kill_chain_phase.create(
+    phase_name="privilege-escalation", kill_chain_name="mitre-attack"
+)
+
 # Registry Run Keys / Startup Folder
-ttp2 = opencti_api_client.attack_pattern.read(
-    filters=[{"key": "x_mitre_id", "values": ["T1060"]}]
+ttp2 = opencti_api_client.attack_pattern.create(
+    name="Boot or Logon Autostart Execution: Registry Run Keys / Startup Folder ",
+    description="Adversaries may achieve persistence by adding a program to a startup folder or referencing it with a Registry run key. Adding an entry to the 'run keys' in the Registry or startup folder will cause the program referenced to be executed when a user logs in. These programs will be executed under the context of the user and will have the account's associated permissions level.",
+    x_mitre_id="T1547.001",
+    killChainPhases=[kcp_pe["id"], kcp_p["id"]],
 )
+ttp2 = opencti_api_client.attack_pattern.read(id=ttp2["id"])
 # Create the relation
 ttp2_relation = opencti_api_client.stix_core_relationship.create(
     fromId=incident["id"],
@@ -132,10 +149,17 @@
 )
 observable_refs.append(observable_ttp2["id"])
 
+kcp_c = opencti_api_client.kill_chain_phase.create(
+    phase_name="collection", kill_chain_name="mitre-attack"
+)
 # Data Encrypted
-ttp3 = opencti_api_client.attack_pattern.read(
-    filters=[{"key": "x_mitre_id", "values": ["T1022"]}]
+ttp3 = opencti_api_client.attack_pattern.create(
+    name=" Archive Collected Data",
+    description="An adversary may compress and/or encrypt data that is collected prior to exfiltration. Compressing the data can help to obfuscate the collected data and minimize the amount of data sent over the network. Encryption can be used to hide information that is being exfiltrated from detection or make exfiltration less conspicuous upon inspection by a defender.",
+    x_mitre_id="T1560",
+    killChainPhases=[kcp_c["id"]],
 )
+ttp3 = opencti_api_client.attack_pattern.read(id=ttp3["id"])
 ttp3_relation = opencti_api_client.stix_core_relationship.create(
     fromId=incident["id"],
     toId=ttp3["id"],