Adding sub-domain enrichment option (#586)

internal-enrichment/hygiene/README.md

@@ -1,8 +1,8 @@
 # OpenCTI Hygiene Connector
 
-this is an internal enrichment connector that uses the following external
+This is an internal enrichment connector that uses the following external
 projects to look for oberservable values in the database that you might want to
-delete / decay because they are known to lead to alse-positives when used for
+delete / decay because they are known to lead to false-positives when used for
 detection:
 
 * [misp-warninglists](https://github.com/MISP/misp-warninglists)
@@ -27,13 +27,15 @@ file of OpenCTI.
 
 ## Configuration
 
-| Parameter            	     | Docker envvar                      | Mandatory | Description                                                                                                                           |
-| -------------------------- | ---------------------------------- | --------- | ------------------------------------------------------------------------------------------------------------------------------------- |
-| `warninglists_slow_search` | `HYGIENE_WARNINGLISTS_SLOW_SEARCH` | No        | Enable slow search mode for the warning lists. If true, uses the most appropriate search method. Can be slower. Default: exact match. |
+| Parameter            	     | Docker envvar                      | Mandatory | Description                                                                                                                                                                 |
+| -------------------------- | ---------------------------------- | --------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `warninglists_slow_search` | `HYGIENE_WARNINGLISTS_SLOW_SEARCH` | No        | Enable slow search mode for the warning lists. If true, uses the most appropriate search method. Can be slower. Default: exact match.                                       |
+| -------------------------- | ---------------------------------- | --------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `enrich_subdomains`        | `HYGIENE_ENRICH_SUBDOMAINS`        | No        | Enable enrichment of sub-domains, This option will add "hygiene_parent" label and ext refs of the parent domain to the subdomain, if sub-domain is not found but parent is. |
 
 ## Behavior
 
-1. Adds a `Hygiene` label on items that correspond to a warning list entry.
+1. Adds a `Hygiene` or `Hygiene_parent` label on items that correspond to a warning list entry.
 2. Adds an external reference for every matching warning list.
 3. Sets the score of all related indicators to a value based on the number of
    reported entries (1:15, >=3:10, >=5:5, default:20).

internal-enrichment/hygiene/docker-compose.yml

@@ -13,4 +13,5 @@ services:
       - CONNECTOR_CONFIDENCE_LEVEL=15 # From 0 (Unknown) to 100 (Fully trusted)
       - CONNECTOR_LOG_LEVEL=info
       - HYGIENE_WARNINGLISTS_SLOW_SEARCH=false # Enable warning lists slow search mode
+      - HYGIENE_ENRICH_SUBDOMAINS=false # Enrich subdomains with hygiene_parent label if the parents are found in warninglists
     restart: always

internal-enrichment/hygiene/src/hygiene.py

@@ -3,6 +3,7 @@
 
 from pymispwarninglists import WarningLists
 from pycti import OpenCTIConnectorHelper, get_config_variable
+import tldextract
 
 # At the moment it is not possible to map lists to their upstream path.
 # Thus we need to have our own mapping here.
@@ -93,6 +94,15 @@ def __init__(self):
             )
         )
 
+        self.enrich_subdomains = bool(
+            get_config_variable(
+                "HYGIENE_ENRICH_SUBDOMAINS",
+                ["hygiene", "enrich_subdomains"],
+                config,
+                default=False,
+            )
+        )
+
         self.helper.log_info(f"Warning lists slow search: {warninglists_slow_search}")
 
         self.warninglists = WarningLists(slow_search=warninglists_slow_search)
@@ -102,13 +112,28 @@ def __init__(self):
             value="Hygiene", color="#fc0341"
         )
 
+        if self.enrich_subdomains:
+            self.label_hygiene_parent = self.helper.api.label.create(
+                value="Hygiene_parent", color="#fc0341"
+            )
+
     def _process_observable(self, observable) -> str:
         # Extract IPv4, IPv6 and Domain from entity data
         observable_value = observable["observable_value"]
+        observable_type = observable["entity_type"]
 
         # Search in warninglist
         result = self.warninglists.search(observable_value)
 
+        # If not found and the domain is a subdomain, search with the parent.
+        use_parent = False
+        if not result and self.enrich_subdomains == True:
+            if observable_type == "Domain-Name":
+                ext = tldextract.extract(observable_value)
+                if observable_value != ext.domain + "." + ext.suffix:
+                    result = self.warninglists.search(ext.domain + "." + ext.suffix)
+                    use_parent = True
+
         # Iterate over the hits
         if result:
             self.helper.log_info(
@@ -135,15 +160,21 @@ def _process_observable(self, observable) -> str:
                     f"number of hits ({len(result)}) setting score to {score}"
                 )
                 self.helper.api.stix_cyber_observable.add_label(
-                    id=observable["id"], label_id=self.label_hygiene["id"]
+                    id=observable["id"],
+                    label_id=self.label_hygiene["id"]
+                    if use_parent == False
+                    else self.label_hygiene_parent["id"],
                 )
                 self.helper.api.stix_cyber_observable.update_field(
                     id=observable["id"],
                     input={"key": "x_opencti_score", "value": score},
                 )
                 for indicator_id in observable["indicatorsIds"]:
                     self.helper.api.stix_domain_object.add_label(
-                        id=indicator_id, label_id=self.label_hygiene["id"]
+                        id=indicator_id,
+                        label_id=self.label_hygiene["id"]
+                        if use_parent == False
+                        else self.label_hygiene_parent["id"],
                     )
                     self.helper.api.stix_domain_object.update_field(
                         id=indicator_id,

internal-enrichment/hygiene/src/requirements.txt

@@ -1,2 +1,3 @@
+tldextract==3.1.2
 pycti==5.1.3
 git+git://github.com/MISP/PyMISPWarningLists.git@main#egg=pymispwarninglists