feat: automated migration

Powlinett · mariot · commit f85a29fb2bc2 · 2025-12-23T15:14:55.000+01:00
diff --git a/internal-enrichment/domaintools/__metadata__/connector_config_schema.json b/internal-enrichment/domaintools/__metadata__/connector_config_schema.json
@@ -0,0 +1,75 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://www.filigran.io/connectors/domaintools_config.schema.json",
+  "type": "object",
+  "properties": {
+    "OPENCTI_URL": {
+      "description": "The base URL of the OpenCTI instance.",
+      "format": "uri",
+      "maxLength": 2083,
+      "minLength": 1,
+      "type": "string"
+    },
+    "OPENCTI_TOKEN": {
+      "description": "The API token to connect to OpenCTI.",
+      "type": "string"
+    },
+    "CONNECTOR_NAME": {
+      "default": "Domaintools",
+      "description": "The name of the connector.",
+      "type": "string"
+    },
+    "CONNECTOR_SCOPE": {
+      "description": "The scope of the connector, e.g. 'flashpoint'.",
+      "items": {
+        "type": "string"
+      },
+      "type": "array"
+    },
+    "CONNECTOR_LOG_LEVEL": {
+      "default": "error",
+      "description": "The minimum level of logs to display.",
+      "enum": [
+        "debug",
+        "info",
+        "warn",
+        "warning",
+        "error"
+      ],
+      "type": "string"
+    },
+    "CONNECTOR_TYPE": {
+      "const": "INTERNAL_ENRICHMENT",
+      "default": "INTERNAL_ENRICHMENT",
+      "type": "string"
+    },
+    "CONNECTOR_AUTO": {
+      "default": false,
+      "description": "Whether the connector should run automatically when an entity is created or updated.",
+      "type": "boolean"
+    },
+    "DOMAINTOOLS_API_USERNAME": {
+      "default": "ChangeMe",
+      "description": "The username required for the authentication on DomainTools API.",
+      "type": "string"
+    },
+    "DOMAINTOOLS_API_KEY": {
+      "default": "ChangeMe",
+      "description": "The password required for the authentication on DomainTools API.",
+      "format": "password",
+      "type": "string",
+      "writeOnly": true
+    },
+    "DOMAINTOOLS_MAX_TLP": {
+      "default": "TLP:AMBER",
+      "description": "The maximal TLP of the observable being enriched.",
+      "type": "string"
+    }
+  },
+  "required": [
+    "OPENCTI_URL",
+    "OPENCTI_TOKEN",
+    "CONNECTOR_SCOPE"
+  ],
+  "additionalProperties": true
+}
diff --git a/internal-enrichment/domaintools/src/connector/__init__.py b/internal-enrichment/domaintools/src/connector/__init__.py
@@ -1,6 +1,6 @@
-# -*- coding: utf-8 -*-
 """DomainTools connector module."""
 
-from .core import DomainToolsConnector
+from connector.connector import DomainToolsConnector
+from connector.settings import ConnectorSettings
 
-__all__ = ["DomainToolsConnector"]
+__all__ = ["DomainToolsConnector", "ConnectorSettings"]
diff --git a/internal-enrichment/domaintools/src/connector/connector.py b/internal-enrichment/domaintools/src/connector/connector.py
@@ -1,15 +1,13 @@
-# -*- coding: utf-8 -*-
 """DomainTools enrichment module."""
 
 from datetime import datetime
-from pathlib import Path
 from typing import Dict
 
 import domaintools
 import stix2
 import validators
-import yaml
-from pycti import Identity, OpenCTIConnectorHelper, get_config_variable
+from connector.settings import ConnectorSettings
+from pycti import Identity, OpenCTIConnectorHelper
 
 from .builder import DtBuilder
 from .constants import DEFAULT_RISK_SCORE, DOMAIN_FIELDS, EMAIL_FIELDS, EntityType
@@ -21,42 +19,18 @@ class DomainToolsConnector:
     _DEFAULT_AUTHOR = "DomainTools"
     _CONNECTOR_RUN_INTERVAL_SEC = 60 * 60
 
-    def __init__(self):
-        # Instantiate the connector helper from config
-        config_file_path = Path(__file__).parent.parent.resolve() / "config.yml"
-        config = (
-            yaml.load(open(config_file_path, encoding="utf-8"), Loader=yaml.FullLoader)
-            if config_file_path.is_file()
-            else {}
+    def __init__(self, config: ConnectorSettings, helper: OpenCTIConnectorHelper):
+        self.config = config
+        self.helper = helper
+        self.api = domaintools.API(
+            self.config.domaintools.api_username, self.config.domaintools.api_key
         )
-        self.helper = OpenCTIConnectorHelper(config, True)
-
-        # DomainTools
-        api_username = get_config_variable(
-            "DOMAINTOOLS_API_USERNAME",
-            ["domaintools", "api_username"],
-            config,
-        )
-        api_key = get_config_variable(
-            "DOMAINTOOLS_API_KEY",
-            ["domaintools", "api_key"],
-            config,
-        )
-        self.api = domaintools.API(api_username, api_key)
-
-        self.max_tlp = get_config_variable(
-            "DOMAINTOOLS_MAX_TLP", ["domaintools", "max_tlp"], config
-        )
-
+        self.max_tlp = self.config.domaintools.max_tlp
         self.author = stix2.Identity(
             id=Identity.generate_id(self._DEFAULT_AUTHOR, "organization"),
             name=self._DEFAULT_AUTHOR,
             identity_class="organization",
-            description=" DomainTools is a leading provider of Whois and other DNS"
-            " profile data for threat intelligence enrichment."
-            " It is a part of the Datacenter Group (DCL Group SA)."
-            " DomainTools data helps security analysts investigate malicious"
-            " activity on their networks.",
+            description=" DomainTools is a leading provider of Whois and other DNS profile data for threat intelligence enrichment. It is a part of the Datacenter Group (DCL Group SA). DomainTools data helps security analysts investigate malicious activity on their networks.",
             confidence=self.helper.connect_confidence_level,
         )
         self.helper.metric.state("idle")
@@ -102,36 +76,28 @@ def _enrich_domaintools(self, builder, opencti_entity) -> str:
             raise ValueError(
                 f"Entity type of the observable: {opencti_entity['entity_type']} not supported."
             )
-
         for entry in results:
             self.helper.log_info(f"Starting enrichment of domain {entry['domain']}")
-            # Retrieve common properties for all relationships.
             builder.reset_score()
             score = entry.get("domain_risk", {}).get("risk_score", DEFAULT_RISK_SCORE)
             builder.set_score(score)
-            # Get the creation date / expiration date for the validity.
             creation_date = entry.get("create_date", {}).get("value", "")
             expiration_date = entry.get("expiration_date", {}).get("value", "")
             if creation_date != "" and expiration_date != "":
                 creation_date = datetime.strptime(creation_date, "%Y-%m-%d")
             if expiration_date != "":
                 expiration_date = datetime.strptime(expiration_date, "%Y-%m-%d")
-
             if creation_date >= expiration_date:
                 self.helper.log_warning(
                     f"Expiration date {expiration_date} not after creation date {creation_date}, not using dates."
                 )
                 creation_date = ""
                 expiration_date = ""
-
-            # In case of IP enrichment, create the domain as it might not exist.
             domain_source_id = (
                 builder.create_domain(entry["domain"])
                 if opencti_entity["entity_type"] == "IPv4-Addr"
                 else opencti_entity["standard_id"]
             )
-
-            # Get ip
             for ip in entry.get("ip", ()):
                 if "address" in ip:
                     ip_id = builder.link_domain_resolves_to(
@@ -145,21 +111,15 @@ def _enrich_domaintools(self, builder, opencti_entity) -> str:
                     if ip_id is not None:
                         for asn in ip.get("asn", ()):
                             builder.link_ip_belongs_to_asn(
-                                ip_id,
-                                asn["value"],
-                                creation_date,
-                                expiration_date,
+                                ip_id, asn["value"], creation_date, expiration_date
                             )
-
-            # Get domains (name-server / mx)
             for category, description in DOMAIN_FIELDS.items():
                 for values in entry.get(category, ()):
                     if (domain := values["domain"]["value"]) != entry["domain"]:
                         if not validators.domain(domain):
                             self.helper.metric.inc("error_count")
                             self.helper.log_warning(
-                                f"[DomainTools] domain {domain} is not correctly "
-                                "formatted. Skipping."
+                                f"[DomainTools] domain {domain} is not correctly formatted. Skipping."
                             )
                             continue
                         new_domain_id = builder.link_domain_resolves_to(
@@ -170,7 +130,6 @@ def _enrich_domaintools(self, builder, opencti_entity) -> str:
                             expiration_date,
                             description,
                         )
-                        # Add the related ips of the name server to the newly created domain.
                         if new_domain_id is not None:
                             for ip in values.get("ip", ()):
                                 builder.link_domain_resolves_to(
@@ -181,8 +140,6 @@ def _enrich_domaintools(self, builder, opencti_entity) -> str:
                                     expiration_date,
                                     f"{description}-ip",
                                 )
-
-            # Emails
             for category, description in EMAIL_FIELDS.items():
                 emails = (
                     entry.get(category, ())
@@ -197,8 +154,6 @@ def _enrich_domaintools(self, builder, opencti_entity) -> str:
                         expiration_date,
                         description,
                     )
-
-            # Domains of emails
             for domain in entry.get("email_domain", ()):
                 if domain["value"] != entry["domain"]:
                     builder.link_domain_resolves_to(
@@ -209,8 +164,6 @@ def _enrich_domaintools(self, builder, opencti_entity) -> str:
                         expiration_date,
                         "email_domain",
                     )
-
-            # Redirects (red)
             if (red := entry.get("redirect_domain", {}).get("value", "")) not in (
                 domain_source_id,
                 "",
@@ -223,7 +176,6 @@ def _enrich_domaintools(self, builder, opencti_entity) -> str:
                     expiration_date,
                     "redirect",
                 )
-
         if len(builder.bundle) > 1:
             builder.send_bundle()
             self.helper.log_info(
@@ -235,31 +187,24 @@ def _enrich_domaintools(self, builder, opencti_entity) -> str:
     def _process_file(self, stix_objects, opencti_entity):
         self.helper.metric.state("running")
         self.helper.metric.inc("run_count")
-
         builder = DtBuilder(self.helper, self.author, stix_objects)
-
-        # Enrichment using DomainTools API.
         result = self._enrich_domaintools(builder, opencti_entity)
         self.helper.metric.state("idle")
         return result
 
     def _process_message(self, data: Dict):
         opencti_entity = data["enrichment_entity"]
-
-        # Extract TLP
         tlp = "TLP:CLEAR"
         for marking_definition in opencti_entity.get("objectMarking", []):
             if marking_definition["definition_type"] == "TLP":
                 tlp = marking_definition["definition"]
-
         if not OpenCTIConnectorHelper.check_max_tlp(tlp, self.max_tlp):
             raise ValueError(
                 "Do not send any data, TLP of the observable is greater than MAX TLP"
             )
-
         stix_objects = data["stix_objects"]
         return self._process_file(stix_objects, opencti_entity)
 
-    def start(self):
+    def run(self):
         """Start the main loop."""
         self.helper.listen(message_callback=self._process_message)
diff --git a/internal-enrichment/domaintools/src/connector/settings.py b/internal-enrichment/domaintools/src/connector/settings.py
@@ -0,0 +1,48 @@
+from connectors_sdk import (
+    BaseConfigModel,
+    BaseConnectorSettings,
+    BaseInternalEnrichmentConnectorConfig,
+)
+from pydantic import Field, SecretStr
+
+
+class InternalEnrichmentConnectorConfig(BaseInternalEnrichmentConnectorConfig):
+    """
+    Override the `BaseInternalEnrichmentConnectorConfig` to add parameters and/or defaults
+    to the configuration for connectors of type `INTERNAL_ENRICHMENT`.
+    """
+
+    name: str = Field(
+        description="The name of the connector.",
+        default="Domaintools",
+    )
+
+
+class DomaintoolsConfig(BaseConfigModel):
+    """
+    Define parameters and/or defaults for the configuration specific to the `DomaintoolsConnector`.
+    """
+
+    api_username: str = Field(
+        description="The username required for the authentication on DomainTools API.",
+        default="ChangeMe",
+    )
+    api_key: SecretStr = Field(
+        description="The password required for the authentication on DomainTools API.",
+        default="ChangeMe",
+    )
+    max_tlp: str = Field(
+        description="The maximal TLP of the observable being enriched.",
+        default="TLP:AMBER",
+    )
+
+
+class ConnectorSettings(BaseConnectorSettings):
+    """
+    Override `BaseConnectorSettings` to include `InternalEnrichmentConnectorConfig` and `DomaintoolsConfig`.
+    """
+
+    connector: InternalEnrichmentConnectorConfig = Field(
+        default_factory=InternalEnrichmentConnectorConfig
+    )
+    domaintools: DomaintoolsConfig = Field(default_factory=DomaintoolsConfig)
diff --git a/internal-enrichment/domaintools/src/main.py b/internal-enrichment/domaintools/src/main.py
@@ -1,8 +1,26 @@
-# -*- coding: utf-8 -*-
-"""DomainTools connector main file."""
+import traceback
 
-from connector import DomainToolsConnector
+from connector import ConnectorSettings, DomainToolsConnector
+from pycti import OpenCTIConnectorHelper
 
 if __name__ == "__main__":
-    connector = DomainToolsConnector()
-    connector.start()
+    """
+    Entry point of the script
+
+    - traceback.print_exc(): This function prints the traceback of the exception to the standard error (stderr).
+    The traceback includes information about the point in the program where the exception occurred,
+    which is very useful for debugging purposes.
+    - exit(1): effective way to terminate a Python program when an error is encountered.
+    It signals to the operating system and any calling processes that the program did not complete successfully.
+    """
+    try:
+        settings = ConnectorSettings()
+        helper = OpenCTIConnectorHelper(
+            config=settings.to_helper_config(), playbook_compatible=True
+        )
+
+        connector = DomainToolsConnector(config=settings, helper=helper)
+        connector.run()
+    except Exception:
+        traceback.print_exc()
+        exit(1)
diff --git a/internal-enrichment/domaintools/src/requirements.txt b/internal-enrichment/domaintools/src/requirements.txt
@@ -1,3 +1,5 @@
 pycti==6.9.3
 domaintools-api==2.6.0
-validators~=0.35.0
+validators~=0.35.0
+pydantic >=2.8.2, <3
+connectors-sdk @ git+https://github.com/OpenCTI-Platform/connectors.git@master#subdirectory=connectors-sdk
diff --git a/internal-enrichment/domaintools/tests/conftest.py b/internal-enrichment/domaintools/tests/conftest.py
@@ -0,0 +1,4 @@
+import os
+import sys
+
+sys.path.append(os.path.join(os.path.dirname(__file__), "..", "src"))
diff --git a/internal-enrichment/domaintools/tests/test-requirements.txt b/internal-enrichment/domaintools/tests/test-requirements.txt
@@ -0,0 +1,2 @@
+-r ../src/requirements.txt
+pytest==8.4.2
diff --git a/internal-enrichment/domaintools/tests/test_main.py b/internal-enrichment/domaintools/tests/test_main.py
diff --git a/internal-enrichment/domaintools/tests/tests_connector/test_settings.py b/internal-enrichment/domaintools/tests/tests_connector/test_settings.py

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+-r ../src/requirements.txt`
	`2`	`+pytest==8.4.2`