diff --git a/api_app/connectors_manager/connectors/misp.py b/api_app/connectors_manager/connectors/misp.py index 650e67c77a..38acd6ce8b 100644 --- a/api_app/connectors_manager/connectors/misp.py +++ b/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" diff --git a/api_app/connectors_manager/connectors/opencti.py b/api_app/connectors_manager/connectors/opencti.py index 95768df0d8..608f1b0e32 100644 --- a/api_app/connectors_manager/connectors/opencti.py +++ b/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( diff --git a/api_app/connectors_manager/connectors/slack.py b/api_app/connectors_manager/connectors/slack.py index f0cb1be3a1..56b08dcc9e 100644 --- a/api_app/connectors_manager/connectors/slack.py +++ b/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 {} diff --git a/api_app/connectors_manager/connectors/yeti.py b/api_app/connectors_manager/connectors/yeti.py index 656b646841..2cab5bf166 100644 --- a/api_app/connectors_manager/connectors/yeti.py +++ b/api_app/connectors_manager/connectors/yeti.py @@ -2,6 +2,7 @@ # See the file 'LICENSE' for copying permission. import ipaddress +import logging import requests from django.conf import settings @@ -9,12 +10,71 @@ 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: diff --git a/tests/api_app/connectors_manager/unit_tests/connectors/test_misp.py b/tests/api_app/connectors_manager/unit_tests/connectors/test_misp.py index f6eaa3539f..9eb6cab69b 100644 --- a/tests/api_app/connectors_manager/unit_tests/connectors/test_misp.py +++ b/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()) diff --git a/tests/api_app/connectors_manager/unit_tests/connectors/test_opencti.py b/tests/api_app/connectors_manager/unit_tests/connectors/test_opencti.py index 4a2741da5e..c409c9e269 100644 --- a/tests/api_app/connectors_manager/unit_tests/connectors/test_opencti.py +++ b/tests/api_app/connectors_manager/unit_tests/connectors/test_opencti.py @@ -4,6 +4,8 @@ from contextlib import ExitStack from unittest.mock import MagicMock, patch +from django.test import override_settings + from api_app.connectors_manager.connectors.opencti import OpenCTI from tests.api_app.connectors_manager.unit_tests.base_test_class import BaseConnectorTest @@ -275,3 +277,75 @@ def test_success_path_integrity(self): self.assertIn("id", result["observable"]) self.assertIn("report", result) self.assertIn("id", result["report"]) + + @override_settings(STAGE_CI=False, MOCK_CONNECTIONS=False) + def test_opencti_health_check_success(self): + connector = self._setup_connector() + + mock_url_param = MagicMock() + mock_url_param.name = "url_key_name" + mock_url_param.value = "http://opencti.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_verify" + mock_ssl_param.value = "false" + + mock_proxies_param = MagicMock() + mock_proxies_param.name = "proxies" + mock_proxies_param.value = None + + 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_proxies_param, + ] + + with patch("api_app.connectors_manager.connectors.opencti.pycti.OpenCTIApiClient") 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_opencti_health_check_failures(self): + connector = self._setup_connector() + + mock_url_param = MagicMock() + mock_url_param.name = "url_key_name" + mock_url_param.value = "http://opencti.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("OpenCTI Connection Exception"), + patch("api_app.connectors_manager.connectors.opencti.pycti.OpenCTIApiClient") as mock_client_cls, + ): + mock_instance = mock_client_cls.return_value + mock_instance.health_check.side_effect = Exception("Connection refused") + self.assertFalse(connector.health_check()) + + with ( + self.subTest("OpenCTI Reports Unhealthy"), + patch("api_app.connectors_manager.connectors.opencti.pycti.OpenCTIApiClient") as mock_client_cls, + ): + mock_instance = mock_client_cls.return_value + mock_instance.health_check.return_value = False + 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()) diff --git a/tests/api_app/connectors_manager/unit_tests/connectors/test_slack.py b/tests/api_app/connectors_manager/unit_tests/connectors/test_slack.py new file mode 100644 index 0000000000..1f2e36e598 --- /dev/null +++ b/tests/api_app/connectors_manager/unit_tests/connectors/test_slack.py @@ -0,0 +1,72 @@ +# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl +# See the file 'LICENSE' for copying permission. + +from unittest.mock import MagicMock, patch + +from django.test import override_settings + +from api_app.connectors_manager.connectors.slack import Slack +from tests.api_app.connectors_manager.unit_tests.base_test_class import BaseConnectorTest + + +class SlackTestCase(BaseConnectorTest): + connector_class = Slack + + @classmethod + def get_extra_config(cls) -> dict: + return { + "_channel": "ABCD", + "slack_username": "intelowl_bot", + "_token": "mock-token-123", + } + + @staticmethod + def get_mocked_response(): + return [patch("api_app.connectors_manager.connectors.slack.slack_sdk.WebClient")] + + def test_connector_run_execution(self): + self.skipTest("Will implement later") + + @override_settings(STAGE_CI=False, MOCK_CONNECTIONS=False) + def test_slack_health_check_success(self): + connector = self._setup_connector() + + mock_token_param = MagicMock() + mock_token_param.name = "token" + mock_token_param.value = "mock-token-123" + + connector._config = MagicMock() + connector._config.parameters.annotate_configured.return_value.annotate_value_for_user.return_value = [ + mock_token_param + ] + + with patch("api_app.connectors_manager.connectors.slack.slack_sdk.WebClient") as mock_webclient_cls: + mock_instance = mock_webclient_cls.return_value + mock_instance.auth_test.return_value = {"ok": True} + + self.assertTrue(connector.health_check()) + + @override_settings(STAGE_CI=False, MOCK_CONNECTIONS=False) + def test_slack_health_check_failures(self): + connector = self._setup_connector() + + mock_token_param = MagicMock() + mock_token_param.name = "token" + mock_token_param.value = "mock-token-123" + + connector._config = MagicMock() + connector._config.parameters.annotate_configured.return_value.annotate_value_for_user.return_value = [ + mock_token_param + ] + + with ( + self.subTest("Slack Connection/Token Exception"), + patch("api_app.connectors_manager.connectors.slack.slack_sdk.WebClient") as mock_webclient_cls, + ): + mock_instance = mock_webclient_cls.return_value + mock_instance.auth_test.side_effect = Exception("invalid_auth") + self.assertFalse(connector.health_check()) + + with self.subTest("Missing Token Configuration"): + connector._config.parameters.annotate_configured.return_value.annotate_value_for_user.return_value = [] + self.assertFalse(connector.health_check()) diff --git a/tests/api_app/connectors_manager/unit_tests/connectors/test_yeti.py b/tests/api_app/connectors_manager/unit_tests/connectors/test_yeti.py index d1c08b6ef2..1cc8e39984 100644 --- a/tests/api_app/connectors_manager/unit_tests/connectors/test_yeti.py +++ b/tests/api_app/connectors_manager/unit_tests/connectors/test_yeti.py @@ -3,6 +3,8 @@ from unittest.mock import MagicMock, patch +import requests +from django.test import override_settings from requests.exceptions import HTTPError from api_app.connectors_manager.connectors.yeti import YETI @@ -138,3 +140,64 @@ def test_yeti_missing_access_token_raises_exception(self): with self.assertRaisesRegex(ConnectorRunException, "Failed to obtain access token from YETI"): connector.run() + + @override_settings(STAGE_CI=False, MOCK_CONNECTIONS=False) + def test_yeti_health_check_success(self): + connector = self._setup_connector() + + mock_url_param = MagicMock() + mock_url_param.name = "url_key_name" + mock_url_param.value = "http://yeti.local/" + + 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 patch( + "api_app.connectors_manager.connectors.yeti.requests.post", side_effect=mock_yeti_api_flow + ): + self.assertTrue(connector.health_check()) + + @override_settings(STAGE_CI=False, MOCK_CONNECTIONS=False) + def test_yeti_health_check_failures(self): + connector = self._setup_connector() + + mock_url_param = MagicMock() + mock_url_param.name = "url_key_name" + mock_url_param.value = "http://yeti.local/" + + 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("Network Exception"), + patch( + "api_app.connectors_manager.connectors.yeti.requests.post", + side_effect=requests.exceptions.Timeout, + ), + ): + self.assertFalse(connector.health_check()) + + with ( + self.subTest("Authentication Failure"), + patch("api_app.connectors_manager.connectors.yeti.requests.post") as mock_post, + ): + mock_post.return_value = MockResponse({"error": "Unauthorized"}, 401) + 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())