diff --git a/docs/snippets/providers/centreon-snippet-autogenerated.mdx b/docs/snippets/providers/centreon-snippet-autogenerated.mdx index 6684090687..55f1a77a5a 100644 --- a/docs/snippets/providers/centreon-snippet-autogenerated.mdx +++ b/docs/snippets/providers/centreon-snippet-autogenerated.mdx @@ -1,10 +1,14 @@ -{/* This snippet is automatically generated using scripts/docs_render_provider_snippets.py +{/* This snippet is automatically generated using scripts/docs_render_provider_snippets.py Do not edit it manually, as it will be overwritten */} ## Authentication This provider requires authentication. - **host_url**: Centreon Host URL (required: True, sensitive: False) - **api_token**: Centreon API Token (required: True, sensitive: True) +- **verify**: Verify SSL certificates (required: False, sensitive: False) +- **ca_certificate**: CA certificate (PEM) used to verify the Centreon server (required: False, sensitive: True) +- **client_certificate**: Client certificate (PEM) for mutual TLS (required: False, sensitive: True) +- **client_key**: Client private key (PEM) for mutual TLS (required: False, sensitive: True) Certain scopes may be required to perform specific actions or queries via the provider. Below is a summary of relevant scopes and their use cases: - **authenticated**: User is authenticated diff --git a/keep/providers/centreon_provider/centreon_provider.py b/keep/providers/centreon_provider/centreon_provider.py index b2a565ed5d..348deee72d 100644 --- a/keep/providers/centreon_provider/centreon_provider.py +++ b/keep/providers/centreon_provider/centreon_provider.py @@ -2,8 +2,11 @@ Centreon is a class that provides a set of methods to interact with the Centreon API. """ +import contextlib import dataclasses import datetime +import os +import tempfile import pydantic import requests @@ -39,6 +42,50 @@ class CentreonProviderAuthConfig: default=None, ) + verify: bool = dataclasses.field( + default=True, + metadata={ + "required": False, + "description": "Verify SSL certificates", + "hint": "Set to false to allow self-signed certificates", + "sensitive": False, + "type": "switch", + }, + ) + + ca_certificate: str = dataclasses.field( + default="", + metadata={ + "name": "ca_certificate", + "description": "CA certificate (PEM) used to verify the Centreon server", + "required": False, + "sensitive": True, + "type": "file", + }, + ) + + client_certificate: str = dataclasses.field( + default="", + metadata={ + "name": "client_certificate", + "description": "Client certificate (PEM) for mutual TLS", + "required": False, + "sensitive": True, + "type": "file", + }, + ) + + client_key: str = dataclasses.field( + default="", + metadata={ + "name": "client_key", + "description": "Client private key (PEM) for mutual TLS", + "required": False, + "sensitive": True, + "type": "file", + }, + ) + class CentreonProvider(BaseProvider): PROVIDER_DISPLAY_NAME = "Centreon" @@ -93,15 +140,54 @@ def __get_headers(self): "centreon-auth-token": self.authentication_config.api_token, } + @contextlib.contextmanager + def __request_kwargs(self): + """Yield the TLS keyword arguments for a requests call. + + The CA/client cert fields hold PEM content, so they are written to + temporary files for the duration of the request (requests expects paths) + and removed afterwards. + """ + temp_paths = [] + + def _materialize(content: str) -> str: + temp = tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) + temp.write(content) + temp.close() + temp_paths.append(temp.name) + return temp.name + + kwargs = {} + auth = self.authentication_config + if not auth.verify: + kwargs["verify"] = False + elif auth.ca_certificate: + kwargs["verify"] = _materialize(auth.ca_certificate) + if auth.client_certificate and auth.client_key: + kwargs["cert"] = ( + _materialize(auth.client_certificate), + _materialize(auth.client_key), + ) + try: + yield kwargs + finally: + for path in temp_paths: + try: + os.unlink(path) + except OSError: + pass + def validate_scopes(self) -> dict[str, bool | str]: """ Validate the scopes of the provider. """ try: - response = requests.get( - self.__get_url("object=centreon_realtime_hosts&action=list"), - headers=self.__get_headers(), - ) + with self.__request_kwargs() as request_kwargs: + response = requests.get( + self.__get_url("object=centreon_realtime_hosts&action=list"), + headers=self.__get_headers(), + **request_kwargs, + ) if response.ok: scopes = {"authenticated": True} else: @@ -118,7 +204,10 @@ def validate_scopes(self) -> dict[str, bool | str]: def __get_host_status(self) -> list[AlertDto]: try: url = self.__get_url("object=centreon_realtime_hosts&action=list") - response = requests.get(url, headers=self.__get_headers()) + with self.__request_kwargs() as request_kwargs: + response = requests.get( + url, headers=self.__get_headers(), **request_kwargs + ) if not response.ok: self.logger.error( @@ -154,7 +243,10 @@ def __get_host_status(self) -> list[AlertDto]: def __get_service_status(self) -> list[AlertDto]: try: url = self.__get_url("object=centreon_realtime_services&action=list") - response = requests.get(url, headers=self.__get_headers()) + with self.__request_kwargs() as request_kwargs: + response = requests.get( + url, headers=self.__get_headers(), **request_kwargs + ) if not response.ok: self.logger.error( diff --git a/tests/providers/centreon_provider/test_centreon_tls.py b/tests/providers/centreon_provider/test_centreon_tls.py new file mode 100644 index 0000000000..f148c4c3fe --- /dev/null +++ b/tests/providers/centreon_provider/test_centreon_tls.py @@ -0,0 +1,127 @@ +"""Tests for TLS/mTLS options on the Centreon provider (issue #3340). + +The provider talks to an on-prem Centreon over HTTPS, which commonly uses a +self-signed or internal-CA certificate. These tests cover the verify switch and +the CA / client-certificate wiring through to the underlying requests call. +""" + +import os +from unittest.mock import MagicMock, patch + +from keep.contextmanager.contextmanager import ContextManager +from keep.providers.centreon_provider.centreon_provider import CentreonProvider +from keep.providers.models.provider_config import ProviderConfig + +REQUESTS_GET = "keep.providers.centreon_provider.centreon_provider.requests.get" + +CA_PEM = "-----BEGIN CERTIFICATE-----\nca-content\n-----END CERTIFICATE-----\n" +CLIENT_PEM = "-----BEGIN CERTIFICATE-----\nclient-content\n-----END CERTIFICATE-----\n" +KEY_PEM = "-----BEGIN PRIVATE KEY-----\nkey-content\n-----END PRIVATE KEY-----\n" + + +def _make_provider(**auth): + context_manager = ContextManager(tenant_id="test", workflow_id="test") + config = ProviderConfig( + authentication={ + "host_url": "https://centreon.internal", + "api_token": "token", + **auth, + }, + name="test-centreon", + ) + return CentreonProvider(context_manager, "centreon", config) + + +def _capturing_get(): + """A requests.get replacement that records the TLS kwargs and reads the + temporary cert files while they still exist (they are removed after the call).""" + captured = {} + + def fake_get(url, headers=None, **kwargs): + captured["verify"] = kwargs.get("verify", "UNSET") + captured["cert"] = kwargs.get("cert") + verify = kwargs.get("verify") + if isinstance(verify, str): + captured["verify_path"] = verify + with open(verify) as fh: + captured["ca_content"] = fh.read() + cert = kwargs.get("cert") + if cert: + captured["cert_paths"] = cert + with open(cert[0]) as fh: + captured["client_cert_content"] = fh.read() + with open(cert[1]) as fh: + captured["client_key_content"] = fh.read() + response = MagicMock() + response.ok = True + response.status_code = 200 + response.json.return_value = [] + return response + + return fake_get, captured + + +def test_verify_true_by_default(): + provider = _make_provider() + fake_get, captured = _capturing_get() + with patch(REQUESTS_GET, side_effect=fake_get): + provider.validate_scopes() + # requests default verify is True, so no explicit override is passed + assert captured["verify"] == "UNSET" + assert captured["cert"] is None + + +def test_verify_false_skips_tls_verification(): + provider = _make_provider(verify=False) + fake_get, captured = _capturing_get() + with patch(REQUESTS_GET, side_effect=fake_get): + provider.validate_scopes() + assert captured["verify"] is False + + +def test_ca_certificate_passed_as_file_path(): + provider = _make_provider(ca_certificate=CA_PEM) + fake_get, captured = _capturing_get() + with patch(REQUESTS_GET, side_effect=fake_get): + provider.validate_scopes() + assert isinstance(captured["verify"], str) + assert captured["ca_content"] == CA_PEM + + +def test_verify_false_takes_precedence_over_ca_certificate(): + provider = _make_provider(verify=False, ca_certificate=CA_PEM) + fake_get, captured = _capturing_get() + with patch(REQUESTS_GET, side_effect=fake_get): + provider.validate_scopes() + assert captured["verify"] is False + + +def test_client_certificate_enables_mutual_tls(): + provider = _make_provider(client_certificate=CLIENT_PEM, client_key=KEY_PEM) + fake_get, captured = _capturing_get() + with patch(REQUESTS_GET, side_effect=fake_get): + provider.validate_scopes() + assert captured["cert"] is not None + assert captured["client_cert_content"] == CLIENT_PEM + assert captured["client_key_content"] == KEY_PEM + + +def test_client_certificate_without_key_is_ignored(): + provider = _make_provider(client_certificate=CLIENT_PEM) + fake_get, captured = _capturing_get() + with patch(REQUESTS_GET, side_effect=fake_get): + provider.validate_scopes() + assert captured["cert"] is None + + +def test_temp_cert_files_are_cleaned_up(): + provider = _make_provider( + ca_certificate=CA_PEM, client_certificate=CLIENT_PEM, client_key=KEY_PEM + ) + fake_get, captured = _capturing_get() + with patch(REQUESTS_GET, side_effect=fake_get): + provider.validate_scopes() + # paths existed during the call but must be removed afterwards + assert not os.path.exists(captured["verify_path"]) + for path in captured["cert_paths"]: + assert not os.path.exists(path)