[Connectors] Implement health check for connectors

api_app/connectors_manager/connectors/misp.py

@@ -1,6 +1,7 @@
 # This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
 # See the file 'LICENSE' for copying permission.
 
+import logging
 from typing import List
 
 import pymisp
@@ -11,6 +12,8 @@
 from api_app.connectors_manager.classes import Connector
 from api_app.connectors_manager.exceptions import ConnectorRunException
 
+logger = logging.getLogger(__name__)
+
 INTELOWL_MISP_TYPE_MAP = {
     Classification.IP: "ip-src",
     Classification.DOMAIN: "domain",
@@ -113,6 +116,64 @@ def _handle_misp_errors(self, errors):
         else:
             raise ConnectorRunException(f"{errors}{debug_info}")
 
+    def health_check(self, user=None) -> bool:
+        if settings.STAGE_CI or settings.MOCK_CONNECTIONS:
+            return True
+
+        params = self._config.parameters.annotate_configured(self._config, user).annotate_value_for_user(
+            self._config, user
+        )
+
+        url = None
+        key = None
+
+        ssl_check = True
+        self_signed_certificate = False
+
+        for param in params:
+            if param.name == "url_key_name":
+                url = param.value
+            elif param.name == "api_key_name":
+                key = param.value
+            elif param.name == "ssl_check":
+                ssl_check = param.value
+            elif param.name == "self_signed_certificate":
+                self_signed_certificate = param.value
+
+        if not url:
+            logger.info("Healthcheck failed: Missing config url")
+            return False
+        if not key:
+            logger.info("Healthcheck failed: Missing config api key")
+            return False
+
+        ssl_param = (
+            f"{settings.PROJECT_LOCATION}/configuration/misp_ssl.crt"
+            if ssl_check and self_signed_certificate
+            else ssl_check
+        )
+
+        try:
+            misp = pymisp.PyMISP(
+                url=url,
+                key=key,
+                ssl=ssl_param,
+                debug=False,
+                timeout=5,
+            )
+
+            # PyMISP has a property misp_instance_version
+            # that makes a GET request to servers/getVersion
+            # using valid API key and returns the version of
+            # the MISP instance if the connection is successful
+            # Refs: https://pymisp.readthedocs.io/en/latest/modules.html?#pymisp.PyMISP.misp_instance_version
+            misp.misp_instance_version
+            return True
+
+        except Exception as e:
+            logger.info(f"MISP health check failed: {e}")
+            return False
+
     def run(self):
         ssl_param = (
             f"{settings.PROJECT_LOCATION}/configuration/misp_ssl.crt"

api_app/connectors_manager/connectors/opencti.py

@@ -1,6 +1,7 @@
 # This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
 # See the file 'LICENSE' for copying permission.
 
+import logging
 from typing import Dict
 
 import pycti
@@ -11,6 +12,8 @@
 from api_app.choices import Classification
 from api_app.connectors_manager import classes
 
+logger = logging.getLogger(__name__)
+
 INTELOWL_OPENCTI_TYPE_MAP = {
     Classification.IP: {
         "v4": "ipv4-addr",
@@ -180,6 +183,49 @@ def _link_report_entities(self, report_id, observable_id, external_ref_id):
                 id=report_id, stixObjectOrStixRelationshipId=observable_id
             )
 
+    def health_check(self, user=None) -> bool:
+        if settings.STAGE_CI or settings.MOCK_CONNECTIONS:
+            return True
+
+        params = self._config.parameters.annotate_configured(self._config, user).annotate_value_for_user(
+            self._config, user
+        )
+
+        url = None
+        token = None
+        ssl_verify = False
+        proxies = None
+
+        for param in params:
+            if param.name == "url_key_name":
+                url = param.value
+            elif param.name == "api_key_name":
+                token = param.value
+            elif param.name == "ssl_verify":
+                ssl_verify = str(param.value).lower() == "true"
+            elif param.name == "proxies":
+                proxies = param.value
+
+        if not url:
+            logger.info("Healthcheck failed: Missing config url")
+            return False
+        if not token:
+            logger.info("Healthcheck failed: Missing config api key")
+            return False
+
+        try:
+            client = pycti.OpenCTIApiClient(url, token, ssl_verify=ssl_verify, proxies=proxies)
+
+            # pycti has a built-in method (health_check) that
+            # returns boolean True/False based on validity of
+            # API key and reachability of the OpenCTI instance
+            # Ref: https://opencti-python-client.readthedocs.io/en/latest/pycti/pycti.api.opencti_api_client.html#pycti.api.opencti_api_client.OpenCTIApiClient.health_check
+            resp = client.health_check()
+            return resp
+        except Exception as e:
+            logger.info(f"OpenCTI health check failed: {e}")
+            return False
+
     def run(self):
         # Initialize OpenCTI client for this run.
         self.opencti_instance = pycti.OpenCTIApiClient(

api_app/connectors_manager/connectors/slack.py

@@ -1,10 +1,17 @@
+# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
+# See the file 'LICENSE' for copying permission.
+
+import logging
 from typing import Dict
 
 import slack_sdk
+from django.conf import settings
 from slack_sdk.errors import SlackApiError
 
 from api_app.connectors_manager.classes import Connector
 
+logger = logging.getLogger(__name__)
+
 
 class Slack(Connector):
     _channel: str
@@ -31,6 +38,37 @@ def body(self) -> str:
             f"for <{self._job.url}/raw|{self._job.analyzable.name}>"
         )
 
+    def health_check(self, user=None) -> bool:
+        if settings.STAGE_CI or settings.MOCK_CONNECTIONS:
+            return True
+
+        params = self._config.parameters.annotate_configured(self._config, user).annotate_value_for_user(
+            self._config, user
+        )
+        token = None
+        for param in params:
+            if param.name == "token":
+                token = param.value
+                break
+
+        if not token:
+            logger.info("Slack health check failed: Missing token configuration.")
+            return False
+
+        try:
+            client = slack_sdk.WebClient(token=token)
+
+            # slack sdk has a built-in method (auth_test) to
+            # test the authentication and connectivity to Slack
+            # (auth_test returns identity information of the
+            # authenticated user if the token is valid)
+            # Ref: https://docs.slack.dev/tools/python-slack-sdk/reference/#slack_sdk.WebClient.auth_test
+            client.auth_test()
+            return True
+        except Exception as e:
+            logger.info(f"Slack health check failed: {e}")
+            return False
+
     def run(self) -> dict:
         self.client.chat_postMessage(text=f"{self.title}\n{self.body}", channel=self._channel, mrkdwn=True)
         return {}

api_app/connectors_manager/connectors/yeti.py

@@ -2,19 +2,79 @@
 # See the file 'LICENSE' for copying permission.
 
 import ipaddress
+import logging
 
 import requests
 from django.conf import settings
 
 from api_app.connectors_manager import classes
 from api_app.connectors_manager.exceptions import ConnectorRunException
 
+logger = logging.getLogger(__name__)
+
 
 class YETI(classes.Connector):
     verify_ssl: bool
     _url_key_name: str
     _api_key_name: str
 
+    def health_check(self, user=None) -> bool:
+        params = self._config.parameters.annotate_configured(self._config, user).annotate_value_for_user(
+            self._config, user
+        )
+        url = None
+        api_key = None
+
+        for param in params:
+            if param.name == "url_key_name":
+                url = param.value
+            elif param.name == "api_key_name":
+                api_key = param.value
+
+        if not url:
+            logger.info("Healthcheck failed: Missing config url")
+            return False
+        if not api_key:
+            logger.info("Healthcheck failed: Missing config api key")
+            return False
+
+        if settings.STAGE_CI or settings.MOCK_CONNECTIONS:
+            return True
+
+        base_url = url.rstrip("/")
+        auth_url = f"{base_url}/api/v2/auth/api-token"
+
+        auth_headers = {"x-yeti-apikey": api_key, "User-Agent": "IntelOwl"}
+
+        try:
+            verify_ssl = getattr(self, "verify_ssl", False)
+
+            # Posting the API key to YETI's authentication endpoint returns an
+            # access token on success (YETI API v2). A valid access token confirms
+            # that the API key is valid and the YETI instance is reachable.
+            # Ref: https://yeti-platform.io/docs/api/#authentication
+            auth_resp = requests.post(
+                url=auth_url,
+                headers=auth_headers,
+                verify=verify_ssl,
+                timeout=10,
+            )
+            auth_resp.raise_for_status()
+            access_token = auth_resp.json().get("access_token")
+
+            if access_token:
+                return True
+            else:
+                logger.info(f"Healthcheck failed for {self}: No access token in response.")
+                return False
+
+        except requests.RequestException as e:
+            logger.info(f"Healthcheck failed: YETI Auth Request failed for {self}. Error: {e}")
+            return False
+        except Exception as e:
+            logger.exception(f"Unexpected error in YETI health_check: {e}")
+            return False
+
     def run(self):
         # get observable value and type
         if self._job.is_sample:

tests/api_app/connectors_manager/unit_tests/connectors/test_misp.py

@@ -3,6 +3,8 @@
 
 from unittest.mock import MagicMock, patch
 
+from django.test import override_settings
+
 from api_app.connectors_manager.connectors.misp import MISP
 from api_app.connectors_manager.exceptions import ConnectorRunException
 from tests.api_app.connectors_manager.unit_tests.base_test_class import BaseConnectorTest
@@ -143,3 +145,68 @@ def test_misp_initialisation_http_failure_raises_exception(self):
                 connector.run()
 
             self.assertIn("plain HTTP request to an HTTPS port", str(context.exception))
+
+    @override_settings(STAGE_CI=False, MOCK_CONNECTIONS=False)
+    def test_misp_health_check_success(self):
+        connector = self._setup_connector()
+
+        mock_url_param = MagicMock()
+        mock_url_param.name = "url_key_name"
+        mock_url_param.value = "http://misp.test/"
+
+        mock_api_param = MagicMock()
+        mock_api_param.name = "api_key_name"
+        mock_api_param.value = "dummy_api_key"
+
+        mock_ssl_param = MagicMock()
+        mock_ssl_param.name = "ssl_check"
+        mock_ssl_param.value = "false"
+
+        mock_cert_param = MagicMock()
+        mock_cert_param.name = "self_signed_certificate"
+        mock_cert_param.value = ""
+
+        connector._config = MagicMock()
+        connector._config.parameters.annotate_configured.return_value.annotate_value_for_user.return_value = [
+            mock_url_param,
+            mock_api_param,
+            mock_ssl_param,
+            mock_cert_param,
+        ]
+
+        with patch("api_app.connectors_manager.connectors.misp.pymisp.PyMISP") as mock_client_cls:
+            mock_instance = mock_client_cls.return_value
+            mock_instance.health_check.return_value = True
+
+            self.assertTrue(connector.health_check())
+
+    @override_settings(STAGE_CI=False, MOCK_CONNECTIONS=False)
+    def test_misp_health_check_failures(self):
+        connector = self._setup_connector()
+
+        mock_url_param = MagicMock()
+        mock_url_param.name = "url_key_name"
+        mock_url_param.value = "http://misp.test/"
+
+        mock_api_param = MagicMock()
+        mock_api_param.name = "api_key_name"
+        mock_api_param.value = "dummy_api_key"
+
+        connector._config = MagicMock()
+        connector._config.parameters.annotate_configured.return_value.annotate_value_for_user.return_value = [
+            mock_url_param,
+            mock_api_param,
+        ]
+
+        with (
+            self.subTest("MISP Connection Exception"),
+            patch(
+                "api_app.connectors_manager.connectors.misp.pymisp.PyMISP",
+                side_effect=Exception("Connection refused"),
+            ),
+        ):
+            self.assertFalse(connector.health_check())
+
+        with self.subTest("Missing Configuration"):
+            connector._config.parameters.annotate_configured.return_value.annotate_value_for_user.return_value = []
+            self.assertFalse(connector.health_check())