[Sekoia/Crowdstrike/Mandiant/AlienVault/RecordedFuture/CisaKEV] Modification on connector to use the new pycti connector helper scheduler

external-import/alienvault/README.md

@@ -20,23 +20,43 @@ integrated to the global `docker-compose.yml` file of OpenCTI.
 If you are using it independently, remember that the connector will try to connect to
 the RabbitMQ on the port configured in the OpenCTI platform.
 
-## Configuration
-
-The connector can be configured with the following variables:
-
-| Config Parameter                   | Docker env var                                | Default                      | Description                                                                                             |
-|------------------------------------|-----------------------------------------------|------------------------------|---------------------------------------------------------------------------------------------------------|
-| `base_url`                         | `ALIENVAULT_BASE_URL`                         | `https://otx.alienvault.com` | The base URL for the OTX DirectConnect API.                                                             |
-| `api_key`                          | `ALIENVAULT_API_KEY`                          | `ChangeMe`                   | The OTX Key.                                                                                            |
-| `tlp`                              | `ALIENVAULT_TLP`                              | `White`                      | The default TLP marking used if the Pulse does not define TLP.                                          |
-| `create_observables`               | `ALIENVAULT_CREATE_OBSERVABLES`               | `true`                       | If true then observables will be created from Pulse indicators and added to the report.                 |
-| `create_indicators`                | `ALIENVAULT_CREATE_INDICATORS`                | `true`                       | If true then indicators will be created from Pulse indicators and added to the report.                  |
-| `pulse_start_timestamp`            | `ALIENVAULT_PULSE_START_TIMESTAMP`            | `2020-05-01T00:00:00`        | The Pulses modified after this timestamp will be imported. Timestamp in ISO 8601 format, UTC.           |
-| `report_status`                    | `ALIENVAULT_REPORT_STATUS`                    | `New`                        | The status of imported reports in the OpenCTI.                                                          |
-| `report_type`                      | `ALIENVAULT_REPORT_TYPE`                      | `threat-report`              | The type of imported reports in the OpenCTI.                                                            |
-| `guess_malware`                    | `ALIENVAULT_GUESS_MALWARE`                    | `false`                      | The Pulse tags are used to guess (queries malwares in the OpenCTI) malwares related to the given Pulse. |
-| `guess_cve`                        | `ALIENVAULT_GUESS_CVE`                        | `false`                      | The Pulse tags are used to guess (checks whether tag matches (CVE-\d{4}-\d{4,7})) vulnerabilities.      |
-| `excluded_pulse_indicator_types`   | `ALIENVAULT_EXCLUDED_PULSE_INDICATOR_TYPES`   | `FileHash-MD5,FileHash-SHA1` | The Pulse indicator types that will be excluded from the import.                                        |
-| `enable_relationships`             | `ALIENVAULT_ENABLE_RELATIONSHIPS`             | `true`                       | If true then the relationships will be created between SDOs.                                            |
-| `enable_attack_patterns_indicates` | `ALIENVAULT_ENABLE_ATTACK_PATTERNS_INDICATES` | `true`                       | If true then the relationships `indicates` will be created between indicators and attack patterns.      |
-| `interval_sec`                     | `ALIENVAULT_INTERVAL_SEC`                     | `1800`                       | The import interval in seconds.                                                                         |
+### Configuration variables
+
+Below are the parameters you'll need to set for OpenCTI:
+
+| Parameter `OpenCTI` | config.yml | Docker environment variable | Mandatory | Description                                          |
+|---------------------|------------|-----------------------------|-----------|------------------------------------------------------|
+| URL                 | url        | `OPENCTI_URL`               | Yes       | The URL of the OpenCTI platform.                     |
+| Token               | token      | `OPENCTI_TOKEN`             | Yes       | The default admin token set in the OpenCTI platform. |
+
+Below are the parameters you'll need to set for running the connector properly:
+
+| Parameter `Connector` | config.yml          | Docker environment variable   | Default      | Mandatory | Description                                                                                      |
+|-----------------------|---------------------|-------------------------------|--------------|-----------|--------------------------------------------------------------------------------------------------|
+| ID                    | `id`                | `CONNECTOR_ID`                | /            | Yes       | A unique `UUIDv4` identifier for this connector instance.                                        |
+| Name                  | `name`              | `CONNECTOR_NAME`              | `AlienVault` | Yes       | Full name of the connector : `AlienVault`.                                                       |
+| Scope                 | `scope`             | `CONNECTOR_SCOPE`             | `alienvault` | Yes       | Must be `alienvault`, not used in this connector.                                                |
+| Run and Terminate     | `run_and_terminate` | `CONNECTOR_RUN_AND_TERMINATE` | `False`      | No        | Launch the connector once if set to True. Takes 2 available values: `True` or `False`.           |
+| Duration Period       | `duration_period`   | `CONNECTOR_DURATION_PERIOD`   | /            | Yes       | Determines the time interval between each launch of the connector in ISO 8601, ex: `PT30M`.      |
+| Queue Threshold       | `queue_threshold`   | `CONNECTOR_QUEUE_THRESHOLD`   | `500`        | No        | Used to determine the limit (RabbitMQ) in MB at which the connector must go into buffering mode. |
+| Log Level             | `log_level`         | `CONNECTOR_LOG_LEVEL`         | /            | Yes       | Determines the verbosity of the logs. Options are `debug`, `info`, `warn`, or `error`.           |
+
+Below are the parameters you'll need to set for AlienVault connector:
+
+| Parameter `AlienVault`           | config.yml                         | Docker environment variable                   | Default                       | Mandatory | Description                                                                                                                    |
+|----------------------------------|------------------------------------|-----------------------------------------------|-------------------------------|-----------|--------------------------------------------------------------------------------------------------------------------------------|
+| Base Url                         | `base_url`                         | `ALIENVAULT_BASE_URL`                         | `https://otx.alienvault.com`  | Yes       | The base URL for the OTX DirectConnect API.                                                                                    |
+| Api Key                          | `api_key`                          | `ALIENVAULT_API_KEY`                          | `ChangeMe`                    | No        | The OTX Key.                                                                                                                   |
+| TLP                              | `tlp`                              | `ALIENVAULT_TLP`                              | `White`                       | Yes       | The default TLP marking used if the Pulse does not define TLP.                                                                 |
+| Create Observables               | `create_observables`               | `ALIENVAULT_CREATE_OBSERVABLES`               | `True`                        | No        | If true then observables will be created from Pulse indicators and added to the report.                                        |
+| Create Indicators                | `create_indicators`                | `ALIENVAULT_CREATE_INDICATORS`                | `True`                        | No        | If true then indicators will be created from Pulse indicators and added to the report.                                         |
+| Pulse Start Timestamp            | `pulse_start_timestamp`            | `ALIENVAULT_PULSE_START_TIMESTAMP`            | `2020-05-01T00:00:00`         | Yes       | The Pulses modified after this timestamp will be imported. Timestamp in ISO 8601 format, UTC.                                  |
+| Report Status                    | `report_status`                    | `ALIENVAULT_REPORT_STATUS`                    | `New`                         | Yes       | The status of imported reports in the OpenCTI.                                                                                 |
+| Report Type                      | `report_type`                      | `ALIENVAULT_REPORT_TYPE`                      | `threat-report`               | No        | The type of imported reports in the OpenCTI.                                                                                   |
+| Guess Malware                    | `guess_malware`                    | `ALIENVAULT_GUESS_MALWARE`                    | `False`                       | Yes       | The Pulse tags are used to guess (queries malwares in the OpenCTI) malwares related to the given Pulse.                        |
+| Guess CVE                        | `guess_cve`                        | `ALIENVAULT_GUESS_CVE`                        | `False`                       | Yes       | The Pulse tags are used to guess (checks whether tag matches (CVE-\d{4}-\d{4,7})) vulnerabilities.                             |
+| Excluded Pulse Indicator Types   | `excluded_pulse_indicator_types`   | `ALIENVAULT_EXCLUDED_PULSE_INDICATOR_TYPES`   | `FileHash-MD5,FileHash-SHA1`  | Yes       | The Pulse indicator types that will be excluded from the import.                                                               |
+| Enable Relationships             | `enable_relationships`             | `ALIENVAULT_ENABLE_RELATIONSHIPS`             | `True`                        | No        | If true then the relationships will be created between SDOs.                                                                   |
+| Enable Attack Patterns Indicates | `enable_attack_patterns_indicates` | `ALIENVAULT_ENABLE_ATTACK_PATTERNS_INDICATES` | `True`                        | No        | If true then the relationships `indicates` will be created between indicators and attack patterns.                             |
+| Filter Indicators                | `filter_indicators`                | `ALIENVAULT_FILTER_INDICATORS`                | `True`                        | No        | This boolean filters out indicators created before the latest pulse datetime, ensuring only recent indicators are processed.   |
+

external-import/alienvault/docker-compose.yml

@@ -9,6 +9,7 @@ services:
       - CONNECTOR_NAME=AlienVault
       - CONNECTOR_SCOPE=alienvault
       - CONNECTOR_LOG_LEVEL=error
+      - CONNECTOR_DURATION_PERIOD=PT30M # In ISO8601 Format starting with "P" for Period ex: "PT30M" = Period time of 30 minutes
       - ALIENVAULT_BASE_URL=https://otx.alienvault.com
       - ALIENVAULT_API_KEY=ChangeMe
       - ALIENVAULT_TLP=White
@@ -22,5 +23,4 @@ services:
       - ALIENVAULT_EXCLUDED_PULSE_INDICATOR_TYPES=FileHash-MD5,FileHash-SHA1  # Excluded Pulse indicator types.
       - ALIENVAULT_ENABLE_RELATIONSHIPS=true                                  # Enable/Disable relationship creation between SDOs.
       - ALIENVAULT_ENABLE_ATTACK_PATTERNS_INDICATES=false                     # Enable/Disable "indicates" relationships between indicators and attack patterns
-      - ALIENVAULT_INTERVAL_SEC=1800
     restart: always

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

@@ -25,8 +25,11 @@
 class AlienVault:
     """AlienVault connector."""
 
+    _CONFIG_CONNECTOR_NAMESPACE = "connector"
     _CONFIG_NAMESPACE = "alienvault"
 
+    _CONFIG_DURATION_PERIOD = f"{_CONFIG_CONNECTOR_NAMESPACE}.duration_period"
+    _CONFIG_INTERVAL_SEC = f"{_CONFIG_NAMESPACE}.interval_sec"
     _CONFIG_BASE_URL = f"{_CONFIG_NAMESPACE}.base_url"
     _CONFIG_API_KEY = f"{_CONFIG_NAMESPACE}.api_key"
     _CONFIG_TLP = f"{_CONFIG_NAMESPACE}.tlp"
@@ -45,9 +48,6 @@ class AlienVault:
         f"{_CONFIG_NAMESPACE}.enable_attack_patterns_indicates"
     )
     _CONFIG_FILTER_INDICATORS = f"{_CONFIG_NAMESPACE}.filter_indicators"
-    _CONFIG_INTERVAL_SEC = f"{_CONFIG_NAMESPACE}.interval_sec"
-
-    _CONFIG_UPDATE_EXISTING_DATA = "connector.update_existing_data"
 
     _CONFIG_REPORT_STATUS_MAPPING = {
         "new": 0,
@@ -62,8 +62,7 @@ class AlienVault:
     _DEFAULT_REPORT_TYPE = "threat-report"
     _DEFAULT_ENABLE_RELATIONSHIPS = True
     _DEFAULT_ENABLE_ATTACK_PATTERNS_INDICATES = True
-
-    _CONNECTOR_RUN_INTERVAL_SEC = 60
+    _DEFAULT_INTERVAL_SEC = 1800
 
     _STATE_LAST_RUN = "last_run"
 
@@ -149,17 +148,21 @@ def __init__(self) -> None:
         else:
             enable_attack_patterns_indicates = bool(enable_attack_patterns_indicates)
 
-        self.interval_sec = self._get_configuration(
-            config, self._CONFIG_INTERVAL_SEC, is_number=True
-        )
-
-        update_existing_data = bool(
-            self._get_configuration(config, self._CONFIG_UPDATE_EXISTING_DATA)
-        )
-
         # Create OpenCTI connector helper
         self.helper = OpenCTIConnectorHelper(config)
 
+        # Get OpenCTI connector duration period
+        self.duration_period = self._get_configuration(
+            config, self._CONFIG_DURATION_PERIOD
+        )
+        self.interval_sec = self._get_configuration(
+            config,
+            self._CONFIG_INTERVAL_SEC,
+            is_number=True,
+        )
+        if not self.interval_sec:
+            self.interval_sec = self._DEFAULT_INTERVAL_SEC
+
         # Create AlienVault author
         author = self._create_author()
 
@@ -174,7 +177,6 @@ def __init__(self) -> None:
             tlp_marking=tlp_marking,
             create_observables=create_observables,
             create_indicators=create_indicators,
-            update_existing_data=update_existing_data,
             default_latest_timestamp=default_latest_pulse_timestamp,
             report_status=report_status,
             report_type=report_type,
@@ -231,66 +233,55 @@ def _convert_report_status_str_to_report_status_int(cls, report_status: str) ->
         return cls._CONFIG_REPORT_STATUS_MAPPING[report_status.lower()]
 
     def run(self):
+
+        if self.duration_period:
+            self.helper.schedule_iso(
+                message_callback=self.process_message,
+                duration_period=self.duration_period,
+            )
+        else:
+            self.helper.schedule_unit(
+                message_callback=self.process_message,
+                duration_period=self.interval_sec,
+                time_unit=self.helper.TimeUnit.SECONDS,
+            )
+
+    def process_message(self):
         """Run AlienVault connector."""
         self._info("Starting AlienVault connector...")
-        while True:
-            self._info("Running AlienVault connector...")
-            run_interval = self._CONNECTOR_RUN_INTERVAL_SEC
-
-            try:
-                timestamp = self._current_unix_timestamp()
-                current_state = self._load_state()
-
-                self._info("Loaded state: {0}", current_state)
-
-                last_run = self._get_state_value(current_state, self._STATE_LAST_RUN)
-                if self._is_scheduled(last_run, timestamp):
-                    now = datetime.datetime.utcfromtimestamp(timestamp)
-                    friendly_name = "AlienVault run @ " + now.strftime(
-                        "%Y-%m-%d %H:%M:%S"
-                    )
-                    work_id = self.helper.api.work.initiate_work(
-                        self.helper.connect_id, friendly_name
-                    )
-                    pulse_import_state = self.pulse_importer.run(current_state, work_id)
-                    new_state = current_state.copy()
-                    new_state.update(pulse_import_state)
-                    new_state[self._STATE_LAST_RUN] = self._current_unix_timestamp()
-
-                    self._info("Storing new state: {0}", new_state)
-                    self.helper.set_state(new_state)
-                    message = (
-                        "State stored, next run in: "
-                        + str(self._get_interval())
-                        + " seconds"
-                    )
-                    self.helper.api.work.to_processed(work_id, message)
-                    self._info(message)
-                else:
-                    next_run = self._get_interval() - (timestamp - last_run)
-                    run_interval = min(run_interval, next_run)
-
-                    self._info(
-                        "Connector will not run, next run in: {0} seconds", next_run
-                    )
-
-            except (KeyboardInterrupt, SystemExit):
-                self._info("Connector stop")
-                sys.exit(0)
-
-            if self.helper.connect_run_and_terminate:
-                self.helper.log_info("Connector stop")
-                self.helper.force_ping()
-                sys.exit(0)
-
-            self._sleep(delay_sec=run_interval)
 
-    @classmethod
-    def _sleep(cls, delay_sec: Optional[int] = None) -> None:
-        sleep_delay = (
-            delay_sec if delay_sec is not None else cls._CONNECTOR_RUN_INTERVAL_SEC
-        )
-        time.sleep(sleep_delay)
+        try:
+            timestamp = self._current_unix_timestamp()
+            current_state = self._load_state()
+
+            self._info("Loaded state: {0}", current_state)
+
+            now = datetime.datetime.utcfromtimestamp(timestamp)
+            friendly_name = "AlienVault run @ " + now.strftime("%Y-%m-%d %H:%M:%S")
+            work_id = self.helper.api.work.initiate_work(
+                self.helper.connect_id, friendly_name
+            )
+            pulse_import_state = self.pulse_importer.run(current_state, work_id)
+            new_state = current_state.copy()
+            new_state.update(pulse_import_state)
+            new_state[self._STATE_LAST_RUN] = self._current_unix_timestamp()
+
+            self._info("Storing new state: {0}", new_state)
+            self.helper.set_state(new_state)
+
+            message = (
+                f"{self.helper.connect_name} connector successfully run, storing last_run as "
+                + str(timestamp)
+            )
+            self.helper.api.work.to_processed(work_id, message)
+            self._info(message)
+
+        except (KeyboardInterrupt, SystemExit):
+            self._info("Connector stopping...")
+            sys.exit(0)
+
+        except Exception as e:  # noqa: B902
+            self._error("AlienVault connector internal error: {0}", str(e))
 
     @staticmethod
     def _current_unix_timestamp() -> int:
@@ -310,16 +301,6 @@ def _get_state_value(
             return state.get(key, default)
         return default
 
-    def _is_scheduled(self, last_run: Optional[int], current_time: int) -> bool:
-        if last_run is None:
-            self._info("Connector first run")
-            return True
-        time_diff = current_time - last_run
-        return time_diff >= self._get_interval()
-
-    def _get_interval(self) -> int:
-        return int(self.interval_sec)
-
     def _info(self, msg: str, *args: Any) -> None:
         fmt_msg = msg.format(*args)
         self.helper.log_info(fmt_msg)