[valhalla] Adapt requirements

README.md

@@ -4,15 +4,15 @@
 [![CircleCI](https://circleci.com/gh/OpenCTI-Platform/connectors.svg?style=shield)](https://circleci.com/gh/OpenCTI-Platform/connectors/tree/master)
 [![Slack Status](https://slack.luatix.org/badge.svg)](https://slack.luatix.org)
 
-The following repository is used to store the OpenCTI connectors for the platform integration with other tools and applications. To know how to enable connectors on OpenCTI, please read the [dedicated documentation](https://www.notion.so/luatix/Connectors-4586c588462d4a1fb5e661f2d9837db8).
+The following repository is used to store the OpenCTI connectors for the platform integration with other tools and applications. To know how to enable connectors on OpenCTI, please read the [dedicated documentation](https://luatix.notion.site/Connectors-4586c588462d4a1fb5e661f2d9837db8).
 
 ## Connectors list and statuses
 
-This repository is used to host connectors that are supported by the core development team of OpenCTI. Nevertheless, the community is also developping a lot of connectors, third-parties modules directly linked to OpenCTI. You can find the list of all available connectors and plugins in the [OpenCTI ecosystem dedicated space](https://www.notion.so/luatix/OpenCTI-Ecosystem-868329e9fb734fca89692b2ed6087e76).
+This repository is used to host connectors that are supported by the core development team of OpenCTI. Nevertheless, the community is also developping a lot of connectors, third-parties modules directly linked to OpenCTI. You can find the list of all available connectors and plugins in the [OpenCTI ecosystem dedicated space](https://luatix.notion.site/OpenCTI-Ecosystem-868329e9fb734fca89692b2ed6087e76).
 
 ## Contributing
 
-If you want to help use improve or develop new connector, please check out the **[development documentation for new connectors](https://www.notion.so/Connector-Development-06b2690697404b5ebc6e3556a1385940)**. If you want to make your connector available to the community, **please create a Pull Request on this repository**, then we will integrate it to the CI and in the [OpenCTI ecosystem](https://www.notion.so/luatix/OpenCTI-Ecosystem-868329e9fb734fca89692b2ed6087e76).
+If you want to help use improve or develop new connector, please check out the **[development documentation for new connectors](https://luatix.notion.site/Connector-Development-06b2690697404b5ebc6e3556a1385940)**. If you want to make your connector available to the community, **please create a Pull Request on this repository**, then we will integrate it to the CI and in the [OpenCTI ecosystem](https://luatix.notion.site/OpenCTI-Ecosystem-868329e9fb734fca89692b2ed6087e76).
 
 ## License
 

external-import/valhalla/docker-compose.yml

@@ -9,7 +9,7 @@ services:
       - CONNECTOR_TYPE=EXTERNAL_IMPORT
       - CONNECTOR_NAME=Valhalla
       - CONNECTOR_SCOPE=valhalla
-      - CONNECTOR_CONFIDENCE_LEVEL=15 # From 0 (Unknown) to 100 (Fully trusted)
+      - CONNECTOR_CONFIDENCE_LEVEL=50 # From 0 (Unknown) to 100 (Fully trusted)
       - CONNECTOR_UPDATE_EXISTING_DATA=false
       - CONNECTOR_LOG_LEVEL=info
       - VALHALLA_API_KEY= # Empty key only fetches public/demo information

external-import/valhalla/src/config.yml.sample

@@ -7,7 +7,7 @@ connector:
   type: 'EXTERNAL_IMPORT'
   name: 'Valhalla'
   scope: 'valhalla'
-  confidence_level: 15 # From 0 (Unknown) to 100 (Fully trusted)
+  confidence_level: 50 # From 0 (Unknown) to 100 (Fully trusted)
   update_existing_data: false
   log_level: 'info'
 

external-import/valhalla/src/requirements.txt

@@ -1,3 +1,4 @@
 pycti==5.2.4
-pydantic==1.8.2
-valhallaAPI==0.3.0
+pydantic==1.9.0
+valhallaAPI==0.5.2
+python-dateutil==2.8.2

external-import/valhalla/src/valhalla/core.py

@@ -56,14 +56,10 @@ def __init__(self):
         # If we run without API key we can assume all data is TLP:WHITE else we
         # default to TLP:AMBER to be safe.
         if self.API_KEY == "" or self.API_KEY is None:
-            self.default_marking = self.helper.api.marking_definition.read(
-                id=TLP_WHITE["id"]
-            )
+            self.default_marking = TLP_WHITE
             self.valhalla_client = ValhallaAPI()
         else:
-            self.default_marking = self.helper.api.marking_definition.read(
-                id=TLP_AMBER["id"]
-            )
+            self.default_marking = TLP_AMBER
             self.valhalla_client = ValhallaAPI(api_key=self.API_KEY)
 
         self.knowledge_importer = KnowledgeImporter(
@@ -74,7 +70,7 @@ def __init__(self):
             self.valhalla_client,
         )
 
-    def run(self):
+    def run(self) -> None:
         self.helper.log_info("starting valhalla connector...")
         while True:
             try:
@@ -96,25 +92,36 @@ def run(self):
                 if self._is_scheduled(last_run, current_time) and self._check_version(
                     last_valhalla_version, api_status.version
                 ):
-                    self.helper.log_info("running importers")
-
-                    knowledge_importer_state = self._run_knowledge_importer(
-                        current_state
+                    self.helper.log_info("running valhalla importer")
+
+                    # Announce upcoming work to OpenCTI
+                    friendly_name = (
+                        "Valhalla run @ "
+                        + datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
+                        + " for upstream database version "
+                        + str(api_status.version)
+                    )
+                    work_id = self.helper.api.work.initiate_work(
+                        self.helper.connect_id, friendly_name
                     )
-                    self.helper.log_info("done with running importers")
+
+                    knowledge_importer_state = self.knowledge_importer.run(work_id)
+
+                    self.helper.log_info("done with running valhalla importer")
 
                     new_state = current_state.copy()
                     new_state.update(knowledge_importer_state)
                     new_state[self._STATE_LAST_RUN] = int(datetime.utcnow().timestamp())
                     new_state[self._VALHALLA_LAST_VERSION] = api_status.version
 
                     self.helper.log_info(f"storing new state: {new_state}")
-
                     self.helper.set_state(new_state)
-
                     self.helper.log_info(
                         f"state stored, next run in: {self._get_interval()} seconds"
                     )
+                    self.helper.api.work.to_processed(
+                        work_id, "Valhalla importer finished"
+                    )
                 else:
                     new_interval = self._get_interval() - (current_time - last_run)
                     self.helper.log_info(
@@ -130,11 +137,6 @@ def run(self):
                 self.helper.log_error(str(e))
                 exit(0)
 
-    def _run_knowledge_importer(
-        self, current_state: Mapping[str, Any]
-    ) -> Mapping[str, Any]:
-        return self.knowledge_importer.run(current_state)
-
     def _get_interval(self) -> int:
         return int(self.INTERVAL_SEC)
 

external-import/valhalla/src/valhalla/knowledge.py

@@ -5,11 +5,12 @@
 import requests
 
 from datetime import datetime
-from typing import Any, List, Mapping
+from typing import Any, Mapping
 from urllib.parse import urlparse
+from dateutil.relativedelta import relativedelta
 
 from .models import ApiResponse, StixEnterpriseAttack
-
+from stix2 import Bundle, Identity, Indicator, ExternalReference, Relationship
 from pycti.connector.opencti_connector_helper import OpenCTIConnectorHelper
 
 
@@ -36,24 +37,34 @@ def __init__(
         self.update_data = update_data
         self.default_marking = default_marking
         self.valhalla_client = valhalla_client
-        self.organization = helper.api.identity.create(
+        self.organization = Identity(
             name="Nextron Systems GmbH",
-            type="Organization",
+            identity_class="organization",
             description="THOR APT scanner and Valhalla Yara Rule API Provider",
         )
+        self.bundle_objects = []
 
-    def run(self, state: Mapping[str, Any]) -> Mapping[str, Any]:
+    def run(self, work_id: int) -> Mapping[str, Any]:
         """Run importer."""
-        self.helper.log_info("running Knowledge importer with state: " + str(state))
+
+        self.bundle_objects.append(self.organization)
 
         self._build_attack_group_mapping()
-        self._process_rules()
+        self.process_yara_rules()
+
+        bundle = Bundle(objects=self.bundle_objects, allow_custom=True).serialize()
+
+        self.helper.send_stix2_bundle(
+            bundle,
+            update=True,
+            work_id=work_id,
+        )
 
         state_timestamp = int(datetime.utcnow().timestamp())
         self.helper.log_info("knowledge importer completed")
         return {self._KNOWLEDGE_IMPORTER_STATE: state_timestamp}
 
-    def _process_rules(self) -> None:
+    def process_yara_rules(self) -> None:
         try:
             rules_json = self.valhalla_client.get_rules_json()
             response = ApiResponse.parse_obj(rules_json)
@@ -62,117 +73,82 @@ def _process_rules(self) -> None:
             return None
 
         for yr in response.rules:
-            try:
-                indicator = self.helper.api.indicator.create(
-                    name=yr.name,
-                    description=yr.cti_description,
-                    pattern_type="yara",
-                    pattern=yr.content,
-                    objectMarking=[self.default_marking["id"]],
-                    created_by=self.organization["id"],
-                    valid_from=yr.cti_date,
-                    x_opencti_main_observable_type="File",
-                    x_opencti_score=yr.score,
-                    x_opencti_detection=True,
-                    update=self.update_data,
-                )
-            except Exception as err:
-                self.helper.log_error(f"error creating indicator: {err}")
-
-            self._add_refs_for_id([yr.reference], indicator["id"])
-            self._add_labels_for_indicator(yr.tags, indicator["id"])
-
-    def _add_labels_for_indicator(self, tags: list, indicator_id: str) -> None:
-        for tag in tags:
-            # handle Mitre ATT&CK relation indicator <-> attack-pattern
-            if re.search(r"^T\d{4}$", tag):
-                self._add_attack_pattern_indicator_by_external_id(tag, indicator_id)
-            # handle Mitre ATT&CK group relation indicator <-> intrusion-set
-            if re.search(r"^G\d{4}$", tag):
-                self._add_intrusion_set_indicator_by_external_id(tag, indicator_id)
-
-            # Create Hygiene Label
-            label_valhalla = self.helper.api.label.create(value=tag, color="#46beda")
-            self.helper.api.stix_domain_object.add_label(
-                id=indicator_id, label_id=label_valhalla["id"]
-            )
-
-    def _add_refs_for_id(self, refs: list, obj_id: str) -> None:
-        if refs == {} or obj_id == "":
-            return None
-
-        for ref in refs:
-            if ref == "-":
-                continue
-            try:
-                san_url = urlparse(ref)
-            except Exception:
-                self.helper.log_error(f"error parsing ref url: {ref}")
-                continue
-
-            reference = self.helper.api.external_reference.create(
-                source_name="Nextron Systems Valhalla API",
-                url=san_url.geturl(),
-                description="Rule Reference: " + san_url.geturl(),
-            )
-            self.helper.api.stix_domain_object.add_external_reference(
-                id=obj_id, external_reference_id=reference["id"]
-            )
-
-    def _add_intrusion_set_indicator_by_external_id(
-        self, external_id: str, indicator_id: str
-    ) -> None:
-        intrusion_set_id = self._ATTACK_MAPPING.get(external_id)
-        if intrusion_set_id == "" or intrusion_set_id is None:
-            self.helper.log_info(f"no intrusion_set found for {external_id}")
-            return None
-
-        # Check if the IS is already in OpenCTI
-        cti_intrusion_set = self.helper.api.intrusion_set.read(id=intrusion_set_id)
-
-        if cti_intrusion_set:
-            self.helper.api.stix_core_relationship.create(
-                fromId=indicator_id,
-                toId=cti_intrusion_set["id"],
-                relationship_type="indicates",
-                description="Yara Rule from Valhalla API",
-                created_by=self.organization["id"],
-                confidence=self.confidence_level,
-            )
-        else:
-            self.helper.log_info(
-                f"intrusion set {intrusion_set_id} not found in OpenCTI. "
-                + "Is the mitre connector configured and running?"
-            )
-
-    def _add_attack_pattern_indicator_by_external_id(
-        self, external_id: str, indicator_id: str
-    ) -> None:
-        attack_pattern_id = self._ATTACK_MAPPING.get(external_id)
-        if attack_pattern_id is None or attack_pattern_id == "":
-            self.helper.log_info(f"no attack_pattern found for {external_id}")
-            return None
-
-        cti_attack_pattern = self.helper.api.attack_pattern.read(id=attack_pattern_id)
-
-        if cti_attack_pattern:
-            self.helper.api.stix_core_relationship.create(
-                fromId=indicator_id,
-                toId=cti_attack_pattern["id"],
-                relationship_type="indicates",
-                created_by=self.organization["id"],
-                description="Yara Rule from Valhalla API",
+            # Handle reference URLs supplied by the Valhalla API
+            refs = []
+            if yr.reference is not None and yr.reference != "" and yr.reference != "-":
+                try:
+                    san_url = urlparse(yr.reference)
+                    ref = ExternalReference(
+                        source_name="Nextron Systems Valhalla API",
+                        url=san_url.geturl(),
+                        description="Rule Reference: " + san_url.geturl(),
+                    )
+                    refs.append(ref)
+                except Exception:
+                    self.helper.log_error(f"error parsing ref url: {yr.reference}")
+                    continue
+
+            indicator = Indicator(
+                name=yr.name,
+                description=yr.cti_description,
+                pattern_type="yara",
+                pattern=yr.content,
+                labels=yr.tags,
+                valid_from=yr.cti_date,
+                valid_until=datetime.utcnow() + relativedelta(years=2),
+                object_marking_refs=[self.default_marking],
+                created_by_ref=self.organization,
                 confidence=self.confidence_level,
-            )
-        else:
-            self.helper.log_info(
-                f"attack pattern {attack_pattern_id} not found in OpenCTI. "
-                + "Is the mitre connector configured and running?"
+                external_references=refs,
+                custom_properties={
+                    "x_opencti_main_observable_type": "StixFile",
+                    "x_opencti_detection": True,
+                    "x_opencti_score": yr.score,
+                },
             )
 
-    @staticmethod
-    def _create_filter(key: str, value: str) -> List[Mapping[str, Any]]:
-        return [{"key": key, "values": [value]}]
+            self.bundle_objects.append(indicator)
+
+            # Handle Tags - those include MITRE ATT&CK tags that we want to
+            # create relationships for
+            for tag in yr.tags:
+                # handle Mitre ATT&CK relation indicator <-> attack-pattern
+                if re.search(r"^T\d{4}$", tag):
+                    attack_pattern_id = self._ATTACK_MAPPING.get(tag)
+
+                    if attack_pattern_id is None or attack_pattern_id == "":
+                        self.helper.log_info(f"no attack_pattern found for {tag}")
+                        return None
+
+                    ap_rel = Relationship(
+                        relationship_type="indicates",
+                        source_ref=indicator,
+                        target_ref=attack_pattern_id,
+                        description="Yara Rule from Valhalla API",
+                        created_by_ref=self.organization,
+                        confidence=self.confidence_level,
+                        object_marking_refs=[self.default_marking],
+                    )
+                    self.bundle_objects.append(ap_rel)
+
+                # handle Mitre ATT&CK group relation indicator <-> intrusion-set
+                if re.search(r"^G\d{4}$", tag):
+                    intrusion_set_id = self._ATTACK_MAPPING.get(tag)
+
+                    if intrusion_set_id == "" or intrusion_set_id is None:
+                        self.helper.log_info(f"no intrusion_set found for {tag}")
+                        return None
+
+                    is_rel = Relationship(
+                        relationship_type="indicates",
+                        source_ref=indicator,
+                        target_ref=intrusion_set_id,
+                        description="Yara Rule from Valhalla API",
+                        created_by_ref=self.organization,
+                        confidence=self.confidence_level,
+                        object_marking_refs=[self.default_marking],
+                    )
+                    self.bundle_objects.append(is_rel)
 
     def _build_attack_group_mapping(self) -> None:
         try:

external-import/valhalla/src/valhalla/models.py

@@ -33,8 +33,8 @@ class YaraRule(BaseModel):
     @property
     def cti_date(self) -> str:
         # Valhalla date format: 2020-04-27 13:28:41
-        d = datetime.strptime(self.date, "%Y-%m-%d %H:%M:%S")
-        return d.strftime("%Y-%m-%dT%H:%M:%S+00:00")
+        return datetime.strptime(self.date, "%Y-%m-%d %H:%M:%S")
+        # return d.strftime("%Y-%m-%dT%H:%M:%S+00:00")
 
     @property
     def cti_description(self) -> str:
@@ -63,8 +63,8 @@ class ApiResponse(BaseModel):
     @property
     def cti_date(self) -> str:
         # Valhalla date format: 2020-04-27 13:28:41
-        d = datetime.strptime(self.date, "%Y-%m-%d %H:%M:%S")
-        return d.strftime("%Y-%m-%dT%H:%M:%S+00:00")
+        return datetime.strptime(self.date, "%Y-%m-%d %H:%M:%S")
+        # return d.strftime("%Y-%m-%dT%H:%M:%S+00:00")
 
 
 class ExternalReference(BaseModel):